diff --git a/.changes/next-release/bugfix-s3streamingoutput-59585.json b/.changes/next-release/bugfix-s3streamingoutput-59585.json
new file mode 100644
index 000000000000..e34eeb66de1b
--- /dev/null
+++ b/.changes/next-release/bugfix-s3streamingoutput-59585.json
@@ -0,0 +1,5 @@
+{
+  "type": "bugfix",
+  "category": "s3, streaming output",
+  "description": "Output files created by S3 Select and streaming output commands are now created with owner-only permissions (0600). Existing files are also tightened to 0600 when overwritten."
+}
diff --git a/awscli/customizations/s3events.py b/awscli/customizations/s3events.py
index 2a0c31d307a7..2cd57fef7975 100644
--- a/awscli/customizations/s3events.py
+++ b/awscli/customizations/s3events.py
@@ -12,6 +12,8 @@
 # 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'
@@ -59,8 +61,7 @@ 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(
-                'Could not find the "output" section for the command: %s'
-                % help_command
+                f'Could not find the "output" section for the command: {help_command}'
             )
     doc.write('======\nOutput\n======\n')
     doc.write(
@@ -98,7 +99,7 @@ class S3SelectStreamOutputArgument(CustomArgument):
     _DOCUMENT_AS_REQUIRED = True
 
     def __init__(self, stream_key, session, **kwargs):
-        super(S3SelectStreamOutputArgument, self).__init__(**kwargs)
+        super().__init__(**kwargs)
         # This is the key in the response body where we can find the
         # streamed contents.
         self._stream_key = stream_key
@@ -120,7 +121,11 @@ def save_file(self, parsed, **kwargs):
         if self._stream_key not in parsed:
             return
         event_stream = parsed[self._stream_key]
-        with open(self._output_file, 'wb') as fp:
+        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:
             for event in event_stream:
                 if 'Records' in event:
                     fp.write(event['Records']['Payload'])
diff --git a/awscli/customizations/streamingoutputarg.py b/awscli/customizations/streamingoutputarg.py
index c4decbb6844d..596aab8a3342 100644
--- a/awscli/customizations/streamingoutputarg.py
+++ b/awscli/customizations/streamingoutputarg.py
@@ -10,6 +10,8 @@
 # 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
@@ -92,7 +94,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(
-            'after-call.%s.%s' % (service_id, operation_name), self.save_file
+            f'after-call.{service_id}.{operation_name}', self.save_file
         )
 
     def save_file(self, parsed, **kwargs):
@@ -104,7 +106,11 @@ def save_file(self, parsed, **kwargs):
             return
         body = parsed[self._response_key]
         buffer_size = self._buffer_size
-        with open(self._output_file, 'wb') as fp:
+        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:
             data = body.read(buffer_size)
             while data:
                 fp.write(data)
diff --git a/tests/functional/s3api/test_select_object_content.py b/tests/functional/s3api/test_select_object_content.py
index 04140563fa09..d643d2843e9e 100644
--- a/tests/functional/s3api/test_select_object_content.py
+++ b/tests/functional/s3api/test_select_object_content.py
@@ -15,19 +15,23 @@
 import shutil
 import tempfile
 
-from awscli.testutils import BaseAWSCommandParamsTest, BaseAWSHelpOutputTest
+from awscli.testutils import (
+    BaseAWSCommandParamsTest,
+    BaseAWSHelpOutputTest,
+    skip_if_windows,
+)
 
 
 class TestGetObject(BaseAWSCommandParamsTest):
     prefix = ['s3api', 'select-object-content']
 
     def setUp(self):
-        super(TestGetObject, self).setUp()
+        super().setUp()
         self.parsed_response = {'Payload': self.create_fake_payload()}
         self._tempdir = tempfile.mkdtemp()
 
     def tearDown(self):
-        super(TestGetObject, self).tearDown()
+        super().tearDown()
         shutil.rmtree(self._tempdir)
 
     def create_fake_payload(self):
@@ -82,6 +86,27 @@ 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)
+        self.assertEqual(os.stat(filename).st_mode & 0o777, 0o600)
+
     def test_errors_are_propagated(self):
         self.http_response.status_code = 400
         self.parsed_response = {
diff --git a/tests/functional/test_streaming_output.py b/tests/functional/test_streaming_output.py
index 10aa6c74407e..0a86c0bdc48b 100644
--- a/tests/functional/test_streaming_output.py
+++ b/tests/functional/test_streaming_output.py
@@ -11,17 +11,23 @@
 # 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
+from awscli.testutils import (
+    BaseAWSCommandParamsTest,
+    FileCreator,
+    skip_if_windows,
+)
 
 
 class TestStreamingOutput(BaseAWSCommandParamsTest):
     def setUp(self):
-        super(TestStreamingOutput, self).setUp()
+        super().setUp()
         self.files = FileCreator()
 
     def tearDown(self):
-        super(TestStreamingOutput, self).tearDown()
+        super().tearDown()
         self.files.remove_all()
 
     def test_get_media_streaming_output(self):
@@ -41,3 +47,17 @@ 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)
+        self.assertEqual(os.stat(outpath).st_mode & 0o777, 0o600)