Initial support for IPP / driverless printing

debian/control

@@ -13,9 +13,10 @@ Depends: ${shlibs:Depends}, ${misc:Depends}, python3, python3-pyqt5, python3-pyq
 Description: securedrop client for qubes workstation
 
 Package: securedrop-export
-Architecture: all
-Depends: ${misc:Depends}, python3, udisks2, cups, printer-driver-brlaser, printer-driver-hpcups, system-config-printer, xpp, libcups2, gnome-disk-utility, libreoffice,
- desktop-file-utils, shared-mime-info, libfile-mimeinfo-perl
+Architecture: amd64
+Depends: ${misc:Depends}, python3, udisks2, cups, cups-ipp-utils, printer-driver-brlaser, printer-driver-hpcups,
+ avahi-daemon, system-config-printer, libcups2, gnome-disk-utility, libreoffice,
+ desktop-file-utils, shared-mime-info, libfile-mimeinfo-perl, gir1.2-gtk-4.0
 Description: Submission export scripts for SecureDrop Workstation
  This package provides scripts used by the SecureDrop Qubes Workstation to
  export submissions from the client to external storage, via the sd-export

debian/securedrop-export.install

@@ -2,3 +2,4 @@ export/files/application-x-sd-export.xml usr/share/mime/packages
 export/files/send-to-usb.desktop usr/share/applications
 export/files/sd-logo.png usr/share/securedrop/icons
 export/files/tcrypt.conf etc/udisks2
+export/files/60-securedrop-export.preset usr/lib/systemd/system-preset

debian/setup-venv.sh

@@ -3,7 +3,7 @@
 set -euxo pipefail
 
 NAME=$1
-if [[ $NAME == "client" ]]; then
+if [[ $NAME == "client" || $NAME == "export" ]]; then
     VENV_ARGS="--system-site-packages"
 else
     VENV_ARGS=""

export/files/60-securedrop-export.preset

@@ -0,0 +1,4 @@
+# avahi-daemon so that driverless printers are detected when
+# plugged in via USB, overriding 75-qubes-vm.preset
+
+enable avahi-daemon.service

export/pyproject.toml

@@ -11,6 +11,15 @@ python = "^3.11"
 pexpect = "^4.9.0"
 
 [tool.poetry.group.dev.dependencies]
+# In production these two are installed using a system package
+# so match those versions exactly
+pygobject = [
+    {version = "=3.42.2", python = ">=3.11"}, # bookworm
+]
+pycairo = [
+    {version = "=1.20.1", python = ">=3.11"}, # bookworm
+]
+
 mypy = "^1.13.0"
 types-setuptools = "^75.2.0"
 pytest = "^8.3.3"

export/securedrop_export/print/print_dialog.py

@@ -0,0 +1,48 @@
+import gi
+gi.require_version("Gtk", "4.0")
+from gi.repository import Gtk
+
+import logging
+from securedrop_export.print.status import Status
+from securedrop_export.exceptions import ExportException
+
+logger = logging.getLogger(__name__)
+
+def open_print_dialog(file_to_print):
+    app = PrintDialog(file_to_print)
+    app.run()
+
+
+class PrintDialog(Gtk.Application):
+    def __init__(self, file_to_print):
+        super().__init__(application_id="org.securedrop.PrintDialog")
+        self.file_to_print = file_to_print
+        self.connect("activate", self.on_activate)
+
+    def on_activate(self, app):
+        window = Gtk.Window(application=app)
+        self.dialog = Gtk.PrintUnixDialog.new("Print Document", window)
+        self.dialog.connect("response", self.on_response)
+        self.dialog.show()
+        window.hide()
+
+    def on_response(self, parent_widget, response_id):
+        if response_id == Gtk.ResponseType.OK:
+            print(f"OK")
+            self.dialog.hide()
+            settings = self.dialog.get_settings()
+            printer = self.dialog.get_selected_printer()
+            page_setup = self.dialog.get_page_setup()
+            job = Gtk.PrintJob.new("print job", printer, settings, page_setup)
+            job.set_source_file(self.file_to_print)
+            job.send(self.on_job_complete, user_data=None)
+        elif response_id == Gtk.ResponseType.APPLY: # Preview (if available)
+            pass
+        elif response_id == Gtk.ResponseType.CANCEL:
+            # FIXME should this exist or should it simply cancel and not report errors
+            raise ExportException(sdstatus=Status.ERROR_PRINT, sderror="User canceled dialog")
+
+    def on_job_complete(self, print_job, user_data, error):
+        if error:
+            self.quit()
+            raise ExportException(sdstatus=Status.ERROR_PRINT, sderror=error.message)

export/securedrop_export/print/service.py

@@ -113,62 +113,64 @@ def printer_test(self) -> Status:
         # a success status here
         return Status.PRINT_TEST_PAGE_SUCCESS
 
-    def _wait_for_print(self):
-        """
-        Use lpstat to ensure the job was fully transfered to the printer
-        Return True if print was successful, otherwise throw ExportException.
-        Currently, the handler `handler` is defined in `exceptions.py`.
-        """
-        signal.signal(signal.SIGALRM, handler)
-        signal.alarm(self.printer_wait_timeout)
-        printer_idle_string = f"printer {self.printer_name} is idle"
-        while True:
-            try:
-                logger.info(f"Running lpstat waiting for printer {self.printer_name}")
-                output = subprocess.check_output(["lpstat", "-p", self.printer_name])
-                if printer_idle_string in output.decode("utf-8"):
-                    logger.info("Print completed")
-                    return True
-                else:
-                    time.sleep(5)
-            except subprocess.CalledProcessError:
-                raise ExportException(sdstatus=Status.ERROR_PRINT)
-            except TimeoutException:
-                logger.error(f"Timeout waiting for printer {self.printer_name}")
-                raise ExportException(sdstatus=Status.ERROR_PRINT)
-        return True
-
     def _check_printer_setup(self) -> None:
         """
         Check printer setup.
         Raise ExportException if supported setup is not found.
         """
+        legacy_printers=False
         try:
             logger.info("Searching for printer")
-            output = subprocess.check_output(["sudo", "lpinfo", "-v"])
-            printers = [x for x in output.decode("utf-8").split() if "usb://" in x]
+
+            printers = self._get_printers_ipp()
             if not printers:
-                logger.info("No usb printers connected")
-                raise ExportException(sdstatus=Status.ERROR_PRINTER_NOT_FOUND)
+                # look for legacy printers after no IPP ones are detected
+                printers = self._get_printers_legacy()
+                legacy_printers=True
 
-            supported_printers = [
-                p for p in printers if any(sub in p for sub in self.SUPPORTED_PRINTERS)
-            ]
-            if not supported_printers:
-                logger.info(f"{printers} are unsupported printers")
-                raise ExportException(sdstatus=Status.ERROR_PRINTER_NOT_SUPPORTED)
+            if not printers:
+                logger.info("No supported printers connected")
+                raise ExportException(sdstatus=Status.ERROR_PRINTER_NOT_FOUND)
 
-            if len(supported_printers) > 1:
-                logger.info("Too many usb printers connected")
+            if len(printers) > 1:
+                logger.info("Too many printers connected")
                 raise ExportException(sdstatus=Status.ERROR_MULTIPLE_PRINTERS_FOUND)
 
             printer_uri = printers[0]
-            printer_ppd = self._install_printer_ppd(printer_uri)
-            self._setup_printer(printer_uri, printer_ppd)
+            if legacy_printers:  # IPP printers are auto-detected by the print dialog
+                self._setup_printer(printer_uri)
         except subprocess.CalledProcessError as e:
             logger.error(e)
             raise ExportException(sdstatus=Status.ERROR_UNKNOWN)
 
+    def _get_printers_legacy(self) -> list[str]:
+        logger.info("Searching for legacy printers")
+        output = subprocess.check_output(["sudo", "lpinfo", "-v"])
+        discovered_printers = [x for x in output.decode("utf-8").split() if "usb://" in x]
+
+        supported_printers = [
+            p for p in discovered_printers if any(sub in p for sub in self.SUPPORTED_PRINTERS)
+        ]
+        if not supported_printers:
+            logger.info(f"{discovered_printers} are unsupported printers")
+            raise ExportException(sdstatus=Status.ERROR_PRINTER_NOT_SUPPORTED)
+
+        return supported_printers
+
+    def _get_printers_ipp(self) -> list[str]:
+        logger.info("Searching for IPP printers (driverless)")
+        discovered_printers = subprocess.check_output(
+            ["ippfind"],
+            universal_newlines=True
+        ).split()
+
+        if discovered_printers:
+            logger.debug(f"Found IPP printers: {', '.join(discovered_printers)}")
+        else:
+            logger.debug(f"No IPP were found")
+
+        return discovered_printers
+
     def _get_printer_uri(self) -> str:
         """
         Get the URI via lpinfo. Only accept URIs of supported printers.
@@ -202,6 +204,9 @@ def _get_printer_uri(self) -> str:
         return printer_uri
 
     def _install_printer_ppd(self, uri):
+        """
+        Discovery and installation of PPD driver (for legacy printers)
+        """
         if not any(x in uri for x in self.SUPPORTED_PRINTERS):
             logger.error(f"Cannot install printer ppd for unsupported printer: {uri}")
             raise ExportException(sdstatus=Status.ERROR_PRINTER_NOT_SUPPORTED)
@@ -390,20 +395,8 @@ def _print_file(self, file_to_print: Path):
             logger.error(f"Something went wrong: {file_to_print} not found")
             raise ExportException(sdstatus=Status.ERROR_PRINT)
 
-        logger.info(f"Sending file to printer {self.printer_name}")
-        try:
-            # We can switch to using libreoffice --pt $printer_cups_name
-            # here, and either print directly (headless) or use the GUI
-            subprocess.check_call(
-                ["xpp", "-P", self.printer_name, file_to_print],
-            )
-        except subprocess.CalledProcessError as e:
-            raise ExportException(sdstatus=Status.ERROR_PRINT, sderror=e.output)
-
-        # This is an addition to ensure that the entire print job is transferred over.
-        # If the job is not fully transferred within the timeout window, the user
-        # will see an error message.
-        self._wait_for_print()
+        logger.info("Opening print dialog")
+        open_print_dialog(str(file_to_print))
 
     def check_output_and_stderr(
         self, command: str, error_status: Status, ignore_stderr_startswith=None

pyproject.toml

@@ -10,6 +10,7 @@ license = "AGPLv3+"
 
 [tool.poetry.dependencies]
 python = "^3.11"
+pygobject = "^3.50.0"
 
 [tool.poetry.group.dev.dependencies]
 ruff = "^0.6.4"