tests: Add termination tests for Qubes

Add termination-related tests for Qubes. To achieve this, we need
to make a change to the Qubes isolation provider. More specifically,
we need to make the isolation provider yield control to the caller only
when the disposable qube is up and running.

Qubes does not provide us a solid guarantee to do so, but we've found a
hacky workaround, whereby we wait until the `qrexec-client-vm` process
opens a `/dev/xen` character device. This should happen, in theory, once
the disposable qube is ready, and has sent a `MSG_SERVICE_CONNECT` RPC
message to the caller.

tests/isolation_provider/test_qubes.py

@@ -1,3 +1,5 @@
+import os
+import pathlib
 import signal
 import subprocess
 import time
@@ -9,7 +11,11 @@ from pytest_mock import MockerFixture
 from dangerzone.conversion import errors
 from dangerzone.document import Document
 from dangerzone.isolation_provider.base import IsolationProvider
-from dangerzone.isolation_provider.qubes import Qubes, running_on_qubes
+from dangerzone.isolation_provider.qubes import (
+    Qubes,
+    is_qubes_native_conversion,
+    running_on_qubes,
+)
 
 # XXX Fixtures used in abstract Test class need to be imported regardless
 from .. import (
@@ -20,7 +26,7 @@ from .. import (
     sanitized_text,
     uncommon_text,
 )
-from .base import IsolationProviderTest
+from .base import IsolationProviderTermination, IsolationProviderTest
 
 
 @pytest.fixture
@@ -28,6 +34,38 @@ def provider() -> Qubes:
     return Qubes()
 
 
+class QubesWait(Qubes):
+    """Qubes isolation provider that blocks until the disposable qube has started."""
+
+    def start_doc_to_pixels_proc(self, document: Document) -> subprocess.Popen:
+        # Check every 100ms if the disposable qube has started. Qubes gives us no
+        # way to figure this out, but `qrexec-client-vm` has an interesting
+        # property. It will start a vchan server **only** once the disposable qube
+        # has started (i.e., on MSG_SERVICE_CONNECT) [1]
+        #
+        # While `qrexec-client-vm` does not report this either, we can snoop on its
+        # open file descriptors, and see when it opens the `/dev/xen` character
+        # devices. This is super flaky and probably subject to race conditions, but
+        # since it's test code, we can live with it.
+        #
+        # [1]: https://www.qubes-os.org/doc/qrexec-internals/#domx-invoke-execution-of-qubes-service-qubesservice-in-domy
+        proc = super().start_doc_to_pixels_proc(document)
+        for i in range(300):
+            for p in pathlib.Path(f"/proc/{proc.pid}/fd").iterdir():
+                if str(p.resolve()).startswith("/dev/xen"):
+                    # The `qrexec-client-vm` process has opened a `/dev/xen` character
+                    # device. We can now yield control back to the caller.
+                    return proc
+            time.sleep(0.1)
+
+        raise RuntimeError(f"Disposable qube did not start within 30 seconds")
+
+
+@pytest.fixture
+def provider_wait() -> QubesWait:
+    return QubesWait()
+
+
 @pytest.mark.skipif(not running_on_qubes(), reason="Not on a Qubes system")
 class TestQubes(IsolationProviderTest):
     def test_out_of_ram(
@@ -53,3 +91,14 @@ class TestQubes(IsolationProviderTest):
             doc = Document(sample_doc)
             provider.doc_to_pixels(doc, tmpdir, proc)
             assert provider.get_proc_exception(proc) == errors.QubesQrexecFailed
+
+
+@pytest.mark.skipif(
+    os.environ.get("DUMMY_CONVERSION", False),
+    reason="cannot run for dummy conversions",
+)
+@pytest.mark.skipif(
+    not is_qubes_native_conversion(), reason="Qubes native conversion is not enabled"
+)
+class TestQubesTermination(IsolationProviderTermination):
+    pass