[v2] Handle broken pipe when command output is piped to an early-exiting reader

.changes/next-release/bugfix-s3-62818.json

@@ -0,0 +1,5 @@
+{
+  "type": "bugfix",
+  "category": "s3",
+  "description": "Fix ``aws s3 ls`` reporting an unhandled ``Broken pipe`` error when its output is piped to a command that exits early, such as ``head``. fixes `#5899 <https://github.com/aws/aws-cli/issues/5899>`__"
+}

awscli/errorhandler.py

@@ -11,8 +11,11 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 import argparse
+import io
 import logging
+import os
 import signal
+import sys
 
 from botocore.exceptions import (
     ClientError,
@@ -44,6 +47,10 @@
 LOG = logging.getLogger(__name__)
 
 VALID_ERROR_FORMATS = ['legacy', 'json', 'yaml', 'text', 'table', 'enhanced']
+# Windows has no SIGPIPE, but writing to a closed pipe can still raise a
+# BrokenPipeError there.  Falling back to the POSIX signal number keeps the
+# return code the same on every platform.
+BROKEN_PIPE_RC = 128 + getattr(signal, 'SIGPIPE', 13)
 # Maximum number of items to display inline for collections
 MAX_INLINE_ITEMS = 5
 
@@ -108,6 +115,7 @@ def construct_entry_point_handlers_chain():
         ParamValidationErrorsHandler(),
         PrompterInterruptExceptionHandler(),
         InterruptExceptionHandler(),
+        BrokenPipeExceptionHandler(),
         GeneralExceptionHandler(),
     ]
     return ChainedExceptionHandler(exception_handlers=handlers)
@@ -124,6 +132,7 @@ def construct_cli_error_handlers_chain(session=None):
         NoCredentialsErrorHandler(session),
         PagerErrorHandler(session),
         InterruptExceptionHandler(),
+        BrokenPipeExceptionHandler(session),
         ClientErrorHandler(session),
         GeneralExceptionHandler(session),
     ]
@@ -388,6 +397,40 @@ def _do_handle_exception(self, exception, stdout, stderr, **kwargs):
         return self.RC
 
 
+def _redirect_stdout_to_devnull():
+    """Point the stdout file descriptor at devnull.
+
+    Without this, the interpreter's final flush of stdout during shutdown
+    fails on the closed pipe.  That prints an "Exception ignored ..."
+    traceback and makes Python exit with its own flush failure status
+    instead of the return code reported by the handler.
+    """
+    try:
+        fileno = sys.stdout.fileno()
+    except (AttributeError, ValueError, io.UnsupportedOperation):
+        # stdout has been replaced by an object with no underlying file
+        # descriptor, so there is nothing to redirect.
+        return
+    devnull = os.open(os.devnull, os.O_WRONLY)
+    try:
+        os.dup2(devnull, fileno)
+    finally:
+        os.close(devnull)
+
+
+class BrokenPipeExceptionHandler(FilteredExceptionHandler):
+    EXCEPTIONS_TO_HANDLE = BrokenPipeError
+    RC = BROKEN_PIPE_RC
+
+    def _do_handle_exception(self, exception, stdout, stderr, **kwargs):
+        # A reader that closes the pipe early is a normal way for a command
+        # to end, e.g. "aws s3 ls s3://bucket/ | head -1".  Unix utilities
+        # exit quietly with 128 + SIGPIPE in this situation, so nothing is
+        # written to stderr here.
+        _redirect_stdout_to_devnull()
+        return self.RC
+
+
 class GeneralExceptionHandler(FilteredExceptionHandler):
     EXCEPTIONS_TO_HANDLE = Exception
     RC = GENERAL_ERROR_RC

awscli/topics/return-codes.rst

@@ -31,6 +31,11 @@ of a CLI command:
 
 * ``130`` -- The process received a SIGINT (Ctrl-C).
 
+* ``141`` -- The command was writing its output to a pipe that was closed
+  before the output was fully written, for example when piping to a command
+  such as ``head`` that exits after reading the lines it needs. This matches
+  the exit status that standard Unix utilities report for a closed pipe.
+
 * ``252`` -- Command syntax was invalid, an unknown parameter was provided, or
   a parameter value was incorrect and prevented the command from running.
 

tests/functional/s3/test_ls_command.py

@@ -11,8 +11,11 @@
 # 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 errno
+
 from dateutil import parser, tz
 
+from awscli.testutils import mock
 from tests.functional.s3 import BaseS3TransferCommandTest
 
 
@@ -417,3 +420,38 @@ def test_list_objects_ignores_bucket_region(self):
         )
         call_args = self.operations_called[0][1]
         self.assertNotIn('BucketRegion', call_args)
+
+
+class TestLSCommandBrokenPipe(BaseS3TransferCommandTest):
+    """Regression tests for aws/aws-cli#5899.
+
+    Piping ``s3 ls`` into a reader that exits early, such as
+    ``aws s3 ls s3://bucket/ | head -1``, closes the pipe while the listing
+    is still being written.  That must end quietly rather than reporting an
+    unhandled error.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.parsed_responses = [
+            {
+                "CommonPrefixes": [],
+                "Contents": [
+                    {
+                        "Key": "foo/bar.txt",
+                        "Size": 100,
+                        "LastModified": "2014-01-09T20:45:49.000Z",
+                    }
+                ],
+            }
+        ]
+
+    def test_closed_pipe_exits_with_sigpipe_rc_and_no_error(self):
+        with mock.patch(
+            'awscli.customizations.s3.subcommands.uni_print',
+            side_effect=BrokenPipeError(errno.EPIPE, 'Broken pipe'),
+        ):
+            _, stderr, _ = self.run_cmd(
+                's3 ls s3://bucket/', expected_rc=128 + 13
+            )
+        self.assertEqual(stderr, '')

tests/unit/test_errorhandler.py

@@ -10,7 +10,9 @@
 # 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 errno
 import io
+import signal
 from collections import namedtuple
 
 import pytest
@@ -32,9 +34,12 @@
     ConfigurationError,
     ParamValidationError,
 )
+from awscli.testutils import mock
 
 Case = namedtuple('Case', ['exception', 'rc', 'stderr', 'stdout'])
 
+BROKEN_PIPE_RC = 128 + 13
+
 
 def _assert_rc_and_error_message(case, error_handler):
     stderr = io.StringIO()
@@ -127,3 +132,88 @@ def test_cli_error_handling_chain_injection(case):
 def test_entry_point_error_handling_chain(case):
     error_handler = errorhandler.construct_entry_point_handlers_chain()
     _assert_rc_and_error_message(case, error_handler)
+
+
+@pytest.fixture
+def broken_pipe_error():
+    return BrokenPipeError(errno.EPIPE, 'Broken pipe')
+
+
+@pytest.fixture
+def no_stdout_redirect():
+    # The handler replaces the process wide stdout file descriptor, which
+    # would swallow the output of the test runner itself.  Tests that care
+    # about the redirect exercise it directly instead.
+    with mock.patch.object(
+        errorhandler, '_redirect_stdout_to_devnull'
+    ) as patched:
+        yield patched
+
+
+@pytest.mark.parametrize(
+    "chain_factory",
+    [
+        errorhandler.construct_entry_point_handlers_chain,
+        errorhandler.construct_cli_error_handlers_chain,
+    ],
+)
+def test_broken_pipe_is_handled_quietly(
+    chain_factory, broken_pipe_error, no_stdout_redirect
+):
+    # A closed downstream pipe is a normal way for a command to end, so it
+    # should not produce any error output.  See aws/aws-cli#5899.
+    stdout = io.StringIO()
+    stderr = io.StringIO()
+
+    rc = chain_factory().handle_exception(broken_pipe_error, stdout, stderr)
+
+    assert rc == BROKEN_PIPE_RC
+    assert stderr.getvalue() == ''
+    assert stdout.getvalue() == ''
+
+
+def test_broken_pipe_rc_matches_sigpipe():
+    # 128 + SIGPIPE is what standard Unix utilities report for a closed pipe.
+    assert errorhandler.BROKEN_PIPE_RC == 128 + signal.SIGPIPE
+
+
+def test_broken_pipe_redirects_stdout(broken_pipe_error, no_stdout_redirect):
+    errorhandler.BrokenPipeExceptionHandler().handle_exception(
+        broken_pipe_error, io.StringIO(), io.StringIO()
+    )
+    no_stdout_redirect.assert_called_once_with()
+
+
+def test_unrelated_os_error_still_reported(no_stdout_redirect):
+    # BrokenPipeError is an OSError subclass, so make sure the new handler
+    # does not start silencing other OSErrors.
+    stdout = io.StringIO()
+    stderr = io.StringIO()
+    error = OSError(errno.EACCES, 'Permission denied')
+
+    rc = errorhandler.construct_entry_point_handlers_chain().handle_exception(
+        error, stdout, stderr
+    )
+
+    assert rc == 255
+    assert 'Permission denied' in stderr.getvalue()
+    no_stdout_redirect.assert_not_called()
+
+
+def test_redirect_stdout_to_devnull_discards_writes(tmp_path):
+    # Use a real file as a stand in for stdout so that the file descriptor
+    # belonging to the test runner is never touched.
+    path = tmp_path / 'stdout.txt'
+    with open(path, 'w') as fake_stdout:
+        with mock.patch('sys.stdout', fake_stdout):
+            errorhandler._redirect_stdout_to_devnull()
+        fake_stdout.write('discarded')
+
+    assert path.read_text() == ''
+
+
+def test_redirect_stdout_to_devnull_without_fileno():
+    # capture_output() and friends replace stdout with an in memory stream
+    # that has no file descriptor, which must not raise.
+    with mock.patch('sys.stdout', io.StringIO()):
+        errorhandler._redirect_stdout_to_devnull()