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

@@ -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):

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):

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,14 +151,17 @@ 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)
     action.device = "/dev/sda"
     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):

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(

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 == ""