Revert "Tighten output file permissions (#10196)" (#10215)

This reverts commit 76e36cb.

Author: AndrewAsseily

awscli/customizations/s3events.py

@@ -12,8 +12,6 @@
 # language governing permissions and limitations under the License.
 """Add S3 specific event streaming output arg."""
 
-import os
-
 from awscli.arguments import CustomArgument
 
 STREAM_HELP_TEXT = 'Filename where the records will be saved'
@@ -61,7 +59,8 @@ def replace_event_stream_docs(help_command, **kwargs):
             # This should never happen, but in the rare case that it does
             # we should be raising something with a helpful error message.
             raise DocSectionNotFoundError(
-                f'Could not find the "output" section for the command: {help_command}'
+                'Could not find the "output" section for the command: %s'
+                % help_command
             )
     doc.write('======\nOutput\n======\n')
     doc.write(
@@ -99,7 +98,7 @@ class S3SelectStreamOutputArgument(CustomArgument):
     _DOCUMENT_AS_REQUIRED = True
 
     def __init__(self, stream_key, session, **kwargs):
-        super().__init__(**kwargs)
+        super(S3SelectStreamOutputArgument, self).__init__(**kwargs)
         # This is the key in the response body where we can find the
         # streamed contents.
         self._stream_key = stream_key
@@ -121,11 +120,7 @@ def save_file(self, parsed, **kwargs):
         if self._stream_key not in parsed:
             return
         event_stream = parsed[self._stream_key]
-        fd = os.open(
-            self._output_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600
-        )
-        os.chmod(self._output_file, 0o600)
-        with os.fdopen(fd, 'wb') as fp:
+        with open(self._output_file, 'wb') as fp:
             for event in event_stream:
                 if 'Records' in event:
                     fp.write(event['Records']['Payload'])

awscli/customizations/streamingoutputarg.py

@@ -10,8 +10,6 @@
 # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
-import os
-
 from botocore.model import Shape
 
 from awscli.arguments import BaseCLIArgument
@@ -94,7 +92,7 @@ def add_to_params(self, parameters, value):
         service_id = self._operation_model.service_model.service_id.hyphenize()
         operation_name = self._operation_model.name
         self._session.register(
-            f'after-call.{service_id}.{operation_name}', self.save_file
+            'after-call.%s.%s' % (service_id, operation_name), self.save_file
         )
 
     def save_file(self, parsed, **kwargs):
@@ -106,11 +104,7 @@ def save_file(self, parsed, **kwargs):
             return
         body = parsed[self._response_key]
         buffer_size = self._buffer_size
-        fd = os.open(
-            self._output_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600
-        )
-        os.chmod(self._output_file, 0o600)
-        with os.fdopen(fd, 'wb') as fp:
+        with open(self._output_file, 'wb') as fp:
             data = body.read(buffer_size)
             while data:
                 fp.write(data)

tests/functional/s3api/test_select_object_content.py

@@ -15,23 +15,19 @@
 import shutil
 import tempfile
 
-from awscli.testutils import (
-    BaseAWSCommandParamsTest,
-    BaseAWSHelpOutputTest,
-    skip_if_windows,
-)
+from awscli.testutils import BaseAWSCommandParamsTest, BaseAWSHelpOutputTest
 
 
 class TestGetObject(BaseAWSCommandParamsTest):
     prefix = ['s3api', 'select-object-content']
 
     def setUp(self):
-        super().setUp()
+        super(TestGetObject, self).setUp()
         self.parsed_response = {'Payload': self.create_fake_payload()}
         self._tempdir = tempfile.mkdtemp()
 
     def tearDown(self):
-        super().tearDown()
+        super(TestGetObject, self).tearDown()
         shutil.rmtree(self._tempdir)
 
     def create_fake_payload(self):
@@ -86,28 +82,6 @@ def test_can_stream_to_file(self):
             contents = f.read()
             self.assertEqual(contents, ('a,b,c,d\n' 'e,f,g,h\n'))
 
-    @skip_if_windows('chmod is not supported on Windows')
-    def test_output_file_permissions(self):
-        filename = os.path.join(self._tempdir, 'outfile_perms')
-        cmdline = self.prefix + [
-            '--bucket',
-            'mybucket',
-            '--key',
-            'mykey',
-            '--expression',
-            'SELECT * FROM S3Object',
-            '--expression-type',
-            'SQL',
-            '--input-serialization',
-            '{"CSV": {}}',
-            '--output-serialization',
-            '{"CSV": {}}',
-            filename,
-        ]
-        self.assert_params_for_cmd(cmdline, ignore_params=True)
-        # Mask file type bits to isolate permission bits (rwxrwxrwx)
-        self.assertEqual(os.stat(filename).st_mode & 0o777, 0o600)
-
     def test_errors_are_propagated(self):
         self.http_response.status_code = 400
         self.parsed_response = {

tests/functional/test_streaming_output.py

@@ -11,23 +11,17 @@
 # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
-import os
-
 from awscli.compat import BytesIO
-from awscli.testutils import (
-    BaseAWSCommandParamsTest,
-    FileCreator,
-    skip_if_windows,
-)
+from awscli.testutils import BaseAWSCommandParamsTest, FileCreator
 
 
 class TestStreamingOutput(BaseAWSCommandParamsTest):
     def setUp(self):
-        super().setUp()
+        super(TestStreamingOutput, self).setUp()
         self.files = FileCreator()
 
     def tearDown(self):
-        super().tearDown()
+        super(TestStreamingOutput, self).tearDown()
         self.files.remove_all()
 
     def test_get_media_streaming_output(self):
@@ -47,18 +41,3 @@ def test_get_media_streaming_output(self):
         self.assert_params_for_cmd(cmdline % outpath, params)
         with open(outpath, 'rb') as outfile:
             self.assertEqual(outfile.read(), b'testbody')
-
-    @skip_if_windows('chmod is not supported on Windows')
-    def test_streaming_output_file_permissions(self):
-        cmdline = (
-            'kinesis-video-media get-media --stream-name test-stream '
-            '--start-selector StartSelectorType=EARLIEST %s'
-        )
-        self.parsed_response = {
-            'ContentType': 'video/webm',
-            'Payload': BytesIO(b'testbody'),
-        }
-        outpath = self.files.full_path('outfile')
-        self.assert_params_for_cmd(cmdline % outpath, ignore_params=True)
-        # Mask file type bits to isolate permission bits (rwxrwxrwx)
-        self.assertEqual(os.stat(outpath).st_mode & 0o777, 0o600)