Handle subprocess stderr output in Python 3.9, treat lpadmin warnings as non-fatal

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