From 6fce35b2efa313b3b44d7afd2ee7d253928a8f1e Mon Sep 17 00:00:00 2001 From: Michael Z Date: Thu, 30 Jun 2022 14:21:23 -0400 Subject: [PATCH] Handle subprocess stderr output in Python 3.9 subprocess.check_call does not raise an exception anymore if there's just output on stderr - hence, we move on to its newer API that allows us to capture stderr and analyze it even if the returncode is 0. It also safely ignores lpadmin driver warnings to allow us to continue to use printer drivers with CUPS. --- securedrop_export/export.py | 23 +++++++++++----------- securedrop_export/print/actions.py | 2 ++ tests/disk/test_actions.py | 13 +++++++++---- tests/print/test_actions.py | 10 +++++----- tests/test_export.py | 31 ++++++++++++++++++++++++++---- 5 files changed, 55 insertions(+), 24 deletions(-) diff --git a/securedrop_export/export.py b/securedrop_export/export.py index fee1537..3483809 100755 --- a/securedrop_export/export.py +++ b/securedrop_export/export.py @@ -118,23 +118,24 @@ def exit_gracefully(self, msg, e=False): # the file with another application sys.exit(0) - def safe_check_call(self, command, error_message): + def safe_check_call(self, command, error_message, ignore_stderr_startswith=None): """ Safely wrap subprocess.check_output to ensure we always return 0 and log the error messages """ try: - subprocess.check_call(command) - except subprocess.CalledProcessError as ex: - # ppdc emits warnings which should not be treated as user facing errors - if ( - ex.returncode == 0 - and ex.stderr is not None - and ex.stderr.startswith("ppdc: Warning") - ): - logger.info("Encountered warning: {}".format(ex.output)) + err = subprocess.run(command, check=True, capture_output=True).stderr + # ppdc and lpadmin may emit warnings we are aware of which should not be treated as + # user facing errors + if ignore_stderr_startswith and err.startswith(ignore_stderr_startswith): + logger.info("Encountered warning: {}".format(err.decode("utf-8"))) + elif err == b"": + # Nothing on stderr and returncode is 0, we're good + pass else: - self.exit_gracefully(msg=error_message, e=ex.output) + self.exit_gracefully(msg=error_message, e=err) + except subprocess.CalledProcessError as ex: + self.exit_gracefully(msg=error_message, e=ex.output) class ExportAction(abc.ABC): diff --git a/securedrop_export/print/actions.py b/securedrop_export/print/actions.py index 1235bcc..01ffad5 100644 --- a/securedrop_export/print/actions.py +++ b/securedrop_export/print/actions.py @@ -143,6 +143,7 @@ def install_printer_ppd(self, uri): "/usr/share/cups/model/", ], error_message=ExportStatus.ERROR_PRINTER_DRIVER_UNAVAILABLE.value, + ignore_stderr_startswith=b"ppdc: Warning", ) return printer_ppd @@ -165,6 +166,7 @@ def setup_printer(self, printer_uri, printer_ppd): "allow:user", ], error_message=ExportStatus.ERROR_PRINTER_INSTALL.value, + ignore_stderr_startswith=b"lpadmin: Printer drivers", ) def print_test_page(self): diff --git a/tests/disk/test_actions.py b/tests/disk/test_actions.py index d82da27..9c553a2 100644 --- a/tests/disk/test_actions.py +++ b/tests/disk/test_actions.py @@ -134,11 +134,14 @@ def test_extract_device_name_multiple_part(mocked_call, capsys, mocker): @mock.patch("subprocess.check_output", return_value=SAMPLE_OUTPUT_NO_PART) -@mock.patch("subprocess.check_call", return_value=0) def test_luks_precheck_encrypted_fde(mocked_call, capsys, mocker): submission = export.SDExport("testfile", TEST_CONFIG) action = DiskExportAction(submission) + command_output = mock.MagicMock() + command_output.stderr = b"" + mocker.patch("subprocess.run", return_value=command_output) + expected_message = export.ExportStatus.USB_ENCRYPTED.value mocked_exit = mocker.patch.object(submission, "exit_gracefully", return_value=0) @@ -148,7 +151,6 @@ def test_luks_precheck_encrypted_fde(mocked_call, capsys, mocker): @mock.patch("subprocess.check_output", return_value=SAMPLE_OUTPUT_ONE_PART) -@mock.patch("subprocess.check_call", return_value=0) def test_luks_precheck_encrypted_single_part(mocked_call, capsys, mocker): submission = export.SDExport("testfile", TEST_CONFIG) action = DiskExportAction(submission) @@ -156,6 +158,10 @@ def test_luks_precheck_encrypted_single_part(mocked_call, capsys, mocker): expected_message = export.ExportStatus.USB_ENCRYPTED.value mocked_exit = mocker.patch.object(submission, "exit_gracefully", return_value=0) + command_output = mock.MagicMock() + command_output.stderr = b"" + mocker.patch("subprocess.run", return_value=command_output) + action.check_luks_volume() mocked_exit.assert_called_once_with(expected_message) @@ -178,7 +184,6 @@ def test_luks_precheck_encrypted_multi_part(mocked_call, capsys, mocker): # Output of `lsblk -o TYPE --noheadings DEVICE_NAME` when a drive has multiple # partitions multi_partition_lsblk_output = b"disk\npart\npart\n" - mocker.patch("subprocess.check_call", return_value=0) mocker.patch("subprocess.check_output", return_value=multi_partition_lsblk_output) with pytest.raises(SystemExit): @@ -202,7 +207,7 @@ def test_luks_precheck_encrypted_luks_error(mocked_call, capsys, mocker): single_partition_lsblk_output = b"disk\npart\n" mocker.patch("subprocess.check_output", return_value=single_partition_lsblk_output) mocker.patch( - "subprocess.check_call", side_effect=CalledProcessError(1, "check_call") + "subprocess.run", side_effect=CalledProcessError(1, "run") ) with pytest.raises(SystemExit): diff --git a/tests/print/test_actions.py b/tests/print/test_actions.py index 176871d..caeeb8f 100644 --- a/tests/print/test_actions.py +++ b/tests/print/test_actions.py @@ -80,7 +80,7 @@ def test_is_not_open_office_file(capsys, open_office_paths): assert not action.is_open_office_file(open_office_paths) -@mock.patch("subprocess.check_call") +@mock.patch("subprocess.run") def test_install_printer_ppd_laserjet(mocker): submission = export.SDExport("testfile", TEST_CONFIG) action = PrintExportAction(submission) @@ -90,7 +90,7 @@ def test_install_printer_ppd_laserjet(mocker): assert ppd == "/usr/share/cups/model/hp-laserjet_6l.ppd" -@mock.patch("subprocess.check_call") +@mock.patch("subprocess.run") def test_install_printer_ppd_brother(mocker): submission = export.SDExport("testfile", TEST_CONFIG) action = PrintExportAction(submission) @@ -105,7 +105,7 @@ def test_install_printer_ppd_error_no_driver(mocker): action = PrintExportAction(submission) mocked_exit = mocker.patch.object(submission, "exit_gracefully", return_value=0) mocker.patch( - "subprocess.check_call", side_effect=CalledProcessError(1, "check_call") + "subprocess.run", side_effect=CalledProcessError(1, "run") ) action.install_printer_ppd( @@ -121,7 +121,7 @@ def test_install_printer_ppd_error_not_supported(mocker): action = PrintExportAction(submission) mocked_exit = mocker.patch.object(submission, "exit_gracefully", return_value=0) mocker.patch( - "subprocess.check_call", side_effect=CalledProcessError(1, "check_call") + "subprocess.run", side_effect=CalledProcessError(1, "run") ) action.install_printer_ppd("usb://Not/Supported?serial=A00000A000000") @@ -134,7 +134,7 @@ def test_setup_printer_error(mocker): action = PrintExportAction(submission) mocked_exit = mocker.patch.object(submission, "exit_gracefully", return_value=0) mocker.patch( - "subprocess.check_call", side_effect=CalledProcessError(1, "check_call") + "subprocess.run", side_effect=CalledProcessError(1, "run") ) action.setup_printer( diff --git a/tests/test_export.py b/tests/test_export.py index dfaab86..602a550 100644 --- a/tests/test_export.py +++ b/tests/test_export.py @@ -515,10 +515,33 @@ def test_valid_encryption_config(capsys): def test_safe_check_call(capsys, mocker): submission = export.SDExport("testfile", TEST_CONFIG) submission.safe_check_call(["ls"], "this will work") - mocked_exit = mocker.patch.object(submission, "exit_gracefully", return_value=0) expected_message = "uh oh!!!!" - submission.safe_check_call(["ls", "kjdsfhkdjfh"], expected_message) + with pytest.raises(SystemExit) as sysexit: + submission.safe_check_call(["ls", "kjdsfhkdjfh"], expected_message) + + assert sysexit.value.code == 0 + + captured = capsys.readouterr() + assert captured.err == "{}\n".format(expected_message) + assert captured.out == "" + + # This should work too + submission.safe_check_call( + ["python3", "-c", "import sys;sys.stderr.write('hello')"], + expected_message, + ignore_stderr_startswith=b'hello', + ) - assert mocked_exit.mock_calls[0][2]["msg"] == expected_message - assert mocked_exit.mock_calls[0][2]["e"] is None + with pytest.raises(SystemExit) as sysexit: + submission.safe_check_call( + ["python3", "-c", "import sys;sys.stderr.write('hello\n')"], + expected_message, + ignore_stderr_startswith=b'world', + ) + + assert sysexit.value.code == 0 + + captured = capsys.readouterr() + assert captured.err == "{}\n".format(expected_message) + assert captured.out == ""