Factor out container utilities to separate module

dangerzone/container_utils.py

@@ -0,0 +1,149 @@
+import gzip
+import logging
+import platform
+import shutil
+import subprocess
+from typing import List, Tuple
+
+from . import errors
+from .util import get_resource_path, get_subprocess_startupinfo
+
+CONTAINER_NAME = "dangerzone.rocks/dangerzone"
+
+log = logging.getLogger(__name__)
+
+
+def get_runtime_name() -> str:
+    if platform.system() == "Linux":
+        runtime_name = "podman"
+    else:
+        # Windows, Darwin, and unknown use docker for now, dangerzone-vm eventually
+        runtime_name = "docker"
+    return runtime_name
+
+
+def get_runtime_version() -> Tuple[int, int]:
+    """Get the major/minor parts of the Docker/Podman version.
+
+    Some of the operations we perform in this module rely on some Podman features
+    that are not available across all of our platforms. In order to have a proper
+    fallback, we need to know the Podman version. More specifically, we're fine with
+    just knowing the major and minor version, since writing/installing a full-blown
+    semver parser is an overkill.
+    """
+    # Get the Docker/Podman version, using a Go template.
+    runtime = get_runtime_name()
+    if runtime == "podman":
+        query = "{{.Client.Version}}"
+    else:
+        query = "{{.Server.Version}}"
+
+    cmd = [runtime, "version", "-f", query]
+    try:
+        version = subprocess.run(
+            cmd,
+            startupinfo=get_subprocess_startupinfo(),
+            capture_output=True,
+            check=True,
+        ).stdout.decode()
+    except Exception as e:
+        msg = f"Could not get the version of the {runtime.capitalize()} tool: {e}"
+        raise RuntimeError(msg) from e
+
+    # Parse this version and return the major/minor parts, since we don't need the
+    # rest.
+    try:
+        major, minor, _ = version.split(".", 3)
+        return (int(major), int(minor))
+    except Exception as e:
+        msg = (
+            f"Could not parse the version of the {runtime.capitalize()} tool"
+            f" (found: '{version}') due to the following error: {e}"
+        )
+        raise RuntimeError(msg)
+
+
+def get_runtime() -> str:
+    container_tech = get_runtime_name()
+    runtime = shutil.which(container_tech)
+    if runtime is None:
+        raise errors.NoContainerTechException(container_tech)
+    return runtime
+
+
+def list_image_tags() -> List[str]:
+    """Get the tags of all loaded Dangerzone images.
+
+    This method returns a mapping of image tags to image IDs, for all Dangerzone
+    images. This can be useful when we want to find which are the local image tags,
+    and which image ID does the "latest" tag point to.
+    """
+    return (
+        subprocess.check_output(
+            [
+                get_runtime(),
+                "image",
+                "list",
+                "--format",
+                "{{ .Tag }}",
+                CONTAINER_NAME,
+            ],
+            text=True,
+            startupinfo=get_subprocess_startupinfo(),
+        )
+        .strip()
+        .split()
+    )
+
+
+def delete_image_tag(tag: str) -> None:
+    """Delete a Dangerzone image tag."""
+    name = CONTAINER_NAME + ":" + tag
+    log.warning(f"Deleting old container image: {name}")
+    try:
+        subprocess.check_output(
+            [get_runtime(), "rmi", "--force", name],
+            startupinfo=get_subprocess_startupinfo(),
+        )
+    except Exception as e:
+        log.warning(
+            f"Couldn't delete old container image '{name}', so leaving it there."
+            f" Original error: {e}"
+        )
+
+
+def get_expected_tag() -> str:
+    """Get the tag of the Dangerzone image tarball from the image-id.txt file."""
+    with open(get_resource_path("image-id.txt")) as f:
+        return f.read().strip()
+
+
+def load_image_tarball() -> None:
+    log.info("Installing Dangerzone container image...")
+    p = subprocess.Popen(
+        [get_runtime(), "load"],
+        stdin=subprocess.PIPE,
+        startupinfo=get_subprocess_startupinfo(),
+    )
+
+    chunk_size = 4 << 20
+    compressed_container_path = get_resource_path("container.tar.gz")
+    with gzip.open(compressed_container_path) as f:
+        while True:
+            chunk = f.read(chunk_size)
+            if len(chunk) > 0:
+                if p.stdin:
+                    p.stdin.write(chunk)
+            else:
+                break
+    _, err = p.communicate()
+    if p.returncode < 0:
+        if err:
+            error = err.decode()
+        else:
+            error = "No output"
+        raise errors.ImageInstallationException(
+            f"Could not install container image: {error}"
+        )
+
+    log.info("Successfully installed container image from")

dangerzone/errors.py

@@ -117,3 +117,26 @@ def handle_document_errors(func: F) -> F:
             sys.exit(1)
 
     return cast(F, wrapper)
+
+
+#### Container-related errors
+
+
+class ImageNotPresentException(Exception):
+    pass
+
+
+class ImageInstallationException(Exception):
+    pass
+
+
+class NoContainerTechException(Exception):
+    def __init__(self, container_tech: str) -> None:
+        super().__init__(f"{container_tech} is not installed")
+
+
+class NotAvailableContainerTechException(Exception):
+    def __init__(self, container_tech: str, error: str) -> None:
+        self.error = error
+        self.container_tech = container_tech
+        super().__init__(f"{container_tech} is not available")

dangerzone/gui/main_window.py

@@ -25,10 +25,6 @@ else:
 
 from .. import errors
 from ..document import SAFE_EXTENSION, Document
-from ..isolation_provider.container import (
-    NoContainerTechException,
-    NotAvailableContainerTechException,
-)
 from ..isolation_provider.qubes import is_qubes_native_conversion
 from ..util import format_exception, get_resource_path, get_version
 from .logic import Alert, CollapsibleBox, DangerzoneGui, UpdateDialog
@@ -496,10 +492,10 @@ class WaitingWidgetContainer(WaitingWidget):
 
         try:
             self.dangerzone.isolation_provider.is_available()
-        except NoContainerTechException as e:
+        except errors.NoContainerTechException as e:
             log.error(str(e))
             state = "not_installed"
-        except NotAvailableContainerTechException as e:
+        except errors.NotAvailableContainerTechException as e:
             log.error(str(e))
             state = "not_running"
             error = e.error

dangerzone/isolation_provider/container.py

@@ -1,13 +1,11 @@
-import gzip
-import json
 import logging
 import os
 import platform
 import shlex
-import shutil
 import subprocess
-from typing import Dict, List, Tuple
+from typing import List
 
+from .. import container_utils, errors
 from ..document import Document
 from ..util import get_resource_path, get_subprocess_startupinfo
 from .base import IsolationProvider, terminate_process_group
@@ -26,88 +24,8 @@ else:
 log = logging.getLogger(__name__)
 
 
-class NoContainerTechException(Exception):
-    def __init__(self, container_tech: str) -> None:
-        super().__init__(f"{container_tech} is not installed")
-
-
-class NotAvailableContainerTechException(Exception):
-    def __init__(self, container_tech: str, error: str) -> None:
-        self.error = error
-        self.container_tech = container_tech
-        super().__init__(f"{container_tech} is not available")
-
-
-class ImageNotPresentException(Exception):
-    pass
-
-
-class ImageInstallationException(Exception):
-    pass
-
-
 class Container(IsolationProvider):
     # Name of the dangerzone container
-    CONTAINER_NAME = "dangerzone.rocks/dangerzone"
-
-    @staticmethod
-    def get_runtime_name() -> str:
-        if platform.system() == "Linux":
-            runtime_name = "podman"
-        else:
-            # Windows, Darwin, and unknown use docker for now, dangerzone-vm eventually
-            runtime_name = "docker"
-        return runtime_name
-
-    @staticmethod
-    def get_runtime_version() -> Tuple[int, int]:
-        """Get the major/minor parts of the Docker/Podman version.
-
-        Some of the operations we perform in this module rely on some Podman features
-        that are not available across all of our platforms. In order to have a proper
-        fallback, we need to know the Podman version. More specifically, we're fine with
-        just knowing the major and minor version, since writing/installing a full-blown
-        semver parser is an overkill.
-        """
-        # Get the Docker/Podman version, using a Go template.
-        runtime = Container.get_runtime_name()
-        if runtime == "podman":
-            query = "{{.Client.Version}}"
-        else:
-            query = "{{.Server.Version}}"
-
-        cmd = [runtime, "version", "-f", query]
-        try:
-            version = subprocess.run(
-                cmd,
-                startupinfo=get_subprocess_startupinfo(),
-                capture_output=True,
-                check=True,
-            ).stdout.decode()
-        except Exception as e:
-            msg = f"Could not get the version of the {runtime.capitalize()} tool: {e}"
-            raise RuntimeError(msg) from e
-
-        # Parse this version and return the major/minor parts, since we don't need the
-        # rest.
-        try:
-            major, minor, _ = version.split(".", 3)
-            return (int(major), int(minor))
-        except Exception as e:
-            msg = (
-                f"Could not parse the version of the {runtime.capitalize()} tool"
-                f" (found: '{version}') due to the following error: {e}"
-            )
-            raise RuntimeError(msg)
-
-    @staticmethod
-    def get_runtime() -> str:
-        container_tech = Container.get_runtime_name()
-        runtime = shutil.which(container_tech)
-        if runtime is None:
-            raise NoContainerTechException(container_tech)
-        return runtime
-
     @staticmethod
     def get_runtime_security_args() -> List[str]:
         """Security options applicable to the outer Dangerzone container.
@@ -131,7 +49,7 @@ class Container(IsolationProvider):
           - This particular argument is specified in `start_doc_to_pixels_proc()`, but
             should move here once #748 is merged.
         """
-        if Container.get_runtime_name() == "podman":
+        if container_utils.get_runtime_name() == "podman":
             security_args = ["--log-driver", "none"]
             security_args += ["--security-opt", "no-new-privileges"]
         else:
@@ -155,114 +73,6 @@ class Container(IsolationProvider):
 
         return security_args
 
-    @staticmethod
-    def list_image_tags() -> Dict[str, str]:
-        """Get the tags of all loaded Dangerzone images.
-
-        Perform the following actions:
-        1. Get the tags of any locally available images that match Dangerzone's image
-           name.
-        2. Get the expected image tag from the image-id.txt file.
-           - If this tag is present in the local images, then we can return.
-           - Else, prune the older container images and continue.
-        3. Load the image tarball and make sure it matches the expected tag.
-        """
-        images = json.loads(
-            subprocess.check_output(
-                [
-                    Container.get_runtime(),
-                    "image",
-                    "list",
-                    "--format",
-                    "json",
-                    Container.CONTAINER_NAME,
-                ],
-                text=True,
-                startupinfo=get_subprocess_startupinfo(),
-            )
-        )
-
-        # Grab every image name and associate it with an image ID.
-        tags = {}
-        for image in images:
-            for name in image["Names"]:
-                tag = name.split(":")[1]
-                tags[tag] = image["Id"]
-
-        return tags
-
-    @staticmethod
-    def delete_image_tag(tag: str) -> None:
-        """Delete a Dangerzone image tag."""
-        name = Container.CONTAINER_NAME + ":" + tag
-        log.warning(f"Deleting old container image: {name}")
-        try:
-            subprocess.check_output(
-                [Container.get_runtime(), "rmi", "--force", name],
-                startupinfo=get_subprocess_startupinfo(),
-            )
-        except Exception as e:
-            log.warning(
-                f"Couldn't delete old container image '{name}', so leaving it there."
-                f" Original error: {e}"
-            )
-
-    @staticmethod
-    def add_image_tag(cur_tag: str, new_tag: str) -> None:
-        """Add a tag to an existing Dangerzone image."""
-        cur_image_name = Container.CONTAINER_NAME + ":" + cur_tag
-        new_image_name = Container.CONTAINER_NAME + ":" + new_tag
-        subprocess.check_output(
-            [
-                Container.get_runtime(),
-                "tag",
-                cur_image_name,
-                new_image_name,
-            ],
-            startupinfo=get_subprocess_startupinfo(),
-        )
-
-        log.info(
-            f"Successfully tagged container image '{cur_image_name}' as '{new_image_name}'"
-        )
-
-    @staticmethod
-    def get_expected_tag() -> str:
-        """Get the tag of the Dangerzone image tarball from the image-id.txt file."""
-        with open(get_resource_path("image-id.txt")) as f:
-            return f.read().strip()
-
-    @staticmethod
-    def load_image_tarball() -> None:
-        log.info("Installing Dangerzone container image...")
-        p = subprocess.Popen(
-            [Container.get_runtime(), "load"],
-            stdin=subprocess.PIPE,
-            startupinfo=get_subprocess_startupinfo(),
-        )
-
-        chunk_size = 4 << 20
-        compressed_container_path = get_resource_path("container.tar.gz")
-        with gzip.open(compressed_container_path) as f:
-            while True:
-                chunk = f.read(chunk_size)
-                if len(chunk) > 0:
-                    if p.stdin:
-                        p.stdin.write(chunk)
-                else:
-                    break
-        _, err = p.communicate()
-        if p.returncode < 0:
-            if err:
-                error = err.decode()
-            else:
-                error = "No output"
-            raise ImageInstallationException(
-                f"Could not install container image: {error}"
-            )
-
-        log.info("Successfully installed container image from")
-
     @staticmethod
     def install() -> bool:
         """Install the container image tarball, or verify that it's already installed.
@@ -275,8 +85,8 @@ class Container(IsolationProvider):
            - Else, prune the older container images and continue.
         3. Load the image tarball and make sure it matches the expected tag.
         """
-        old_tags = Container.list_image_tags()
+        old_tags = container_utils.list_image_tags()
-        expected_tag = Container.get_expected_tag()
+        expected_tag = container_utils.get_expected_tag()
 
         if expected_tag not in old_tags:
             # Prune older container images.
@@ -289,14 +99,14 @@ class Container(IsolationProvider):
             return True
 
         # Load the image tarball into the container runtime.
-        Container.load_image_tarball()
+        container_utils.load_image_tarball()
 
         # Check that the container image has the expected image tag.
         # See https://github.com/freedomofpress/dangerzone/issues/988 for an example
         # where this was not the case.
-        new_tags = Container.list_image_tags()
+        new_tags = container_utils.list_image_tags()
         if expected_tag not in new_tags:
-            raise ImageNotPresentException(
+            raise errors.ImageNotPresentException(
                 f"Could not find expected tag '{expected_tag}' after loading the"
                 " container image tarball"
             )
@@ -309,8 +119,8 @@ class Container(IsolationProvider):
 
     @staticmethod
     def is_available() -> bool:
-        container_runtime = Container.get_runtime()
+        container_runtime = container_utils.get_runtime()
-        runtime_name = Container.get_runtime_name()
+        runtime_name = container_utils.get_runtime_name()
         # Can we run `docker/podman image ls` without an error
         with subprocess.Popen(
             [container_runtime, "image", "ls"],
@@ -320,7 +130,9 @@ class Container(IsolationProvider):
         ) as p:
             _, stderr = p.communicate()
             if p.returncode != 0:
-                raise NotAvailableContainerTechException(runtime_name, stderr.decode())
+                raise errors.NotAvailableContainerTechException(
+                    runtime_name, stderr.decode()
+                )
             return True
 
     def doc_to_pixels_container_name(self, document: Document) -> str:
@@ -355,7 +167,7 @@ class Container(IsolationProvider):
         name: str,
         extra_args: List[str] = [],
     ) -> subprocess.Popen:
-        container_runtime = self.get_runtime()
+        container_runtime = container_utils.get_runtime()
         security_args = self.get_runtime_security_args()
         enable_stdin = ["-i"]
         set_name = ["--name", name]
@@ -385,7 +197,7 @@ class Container(IsolationProvider):
         connected to the Docker daemon, and killing it will just close the associated
         standard streams.
         """
-        container_runtime = self.get_runtime()
+        container_runtime = container_utils.get_runtime()
         cmd = [container_runtime, "kill", name]
         try:
             # We do not check the exit code of the process here, since the container may
@@ -421,8 +233,8 @@ class Container(IsolationProvider):
         # NOTE: Using `--userns nomap` is available only on Podman >= 4.1.0.
         # XXX: Move this under `get_runtime_security_args()` once #748 is merged.
         extra_args = []
-        if Container.get_runtime_name() == "podman":
+        if container_utils.get_runtime_name() == "podman":
-            if Container.get_runtime_version() >= (4, 1):
+            if container_utils.get_runtime_version() >= (4, 1):
                 extra_args += ["--userns", "nomap"]
 
         name = self.doc_to_pixels_container_name(document)
@@ -449,7 +261,7 @@ class Container(IsolationProvider):
         # after a podman kill / docker kill invocation, this will likely be the case,
         # else the container runtime (Docker/Podman) has experienced a problem, and we
         # should report it.
-        container_runtime = self.get_runtime()
+        container_runtime = container_utils.get_runtime()
         name = self.doc_to_pixels_container_name(document)
         all_containers = subprocess.run(
             [container_runtime, "ps", "-a"],
@@ -471,11 +283,11 @@ class Container(IsolationProvider):
             if cpu_count is not None:
                 n_cpu = cpu_count
 
-        elif self.get_runtime_name() == "docker":
+        elif container_utils.get_runtime_name() == "docker":
             # For Windows and MacOS containers run in VM
             # So we obtain the CPU count for the VM
             n_cpu_str = subprocess.check_output(
-                [self.get_runtime(), "info", "--format", "{{.NCPU}}"],
+                [container_utils.get_runtime(), "info", "--format", "{{.NCPU}}"],
                 text=True,
                 startupinfo=get_subprocess_startupinfo(),
             )

tests/gui/test_main_window.py

@@ -10,6 +10,7 @@ from pytest_mock import MockerFixture
 from pytest_subprocess import FakeProcess
 from pytestqt.qtbot import QtBot
 
+from dangerzone import errors
 from dangerzone.document import Document
 from dangerzone.gui import MainWindow
 from dangerzone.gui import main_window as main_window_module
@@ -25,11 +26,7 @@ from dangerzone.gui.main_window import (
     WaitingWidgetContainer,
 )
 from dangerzone.gui.updater import UpdateReport, UpdaterThread
-from dangerzone.isolation_provider.container import (
+from dangerzone.isolation_provider.container import Container
-    Container,
-    NoContainerTechException,
-    NotAvailableContainerTechException,
-)
 from dangerzone.isolation_provider.dummy import Dummy
 
 from .test_updater import assert_report_equal, default_updater_settings
@@ -513,7 +510,7 @@ def test_not_available_container_tech_exception(
     mock_app = mocker.MagicMock()
     dummy = Dummy()
     fn = mocker.patch.object(dummy, "is_available")
-    fn.side_effect = NotAvailableContainerTechException(
+    fn.side_effect = errors.NotAvailableContainerTechException(
         "podman", "podman image ls logs"
     )
 
@@ -536,7 +533,7 @@ def test_no_container_tech_exception(qtbot: QtBot, mocker: MockerFixture) -> Non
     dummy = mocker.MagicMock()
 
     # Raise
-    dummy.is_available.side_effect = NoContainerTechException("podman")
+    dummy.is_available.side_effect = errors.NoContainerTechException("podman")
 
     dz = DangerzoneGui(mock_app, dummy)
     widget = WaitingWidgetContainer(dz)

tests/isolation_provider/test_container.py

@@ -4,12 +4,8 @@ import pytest
 from pytest_mock import MockerFixture
 from pytest_subprocess import FakeProcess
 
-from dangerzone.isolation_provider.container import (
+from dangerzone import container_utils, errors
-    Container,
+from dangerzone.isolation_provider.container import Container
-    ImageInstallationException,
-    ImageNotPresentException,
-    NotAvailableContainerTechException,
-)
 from dangerzone.isolation_provider.qubes import is_qubes_native_conversion
 
 from .base import IsolationProviderTermination, IsolationProviderTest
@@ -33,11 +29,11 @@ class TestContainer(IsolationProviderTest):
         the "podman image ls" command fails.
         """
         fp.register_subprocess(
-            [provider.get_runtime(), "image", "ls"],
+            [container_utils.get_runtime(), "image", "ls"],
             returncode=-1,
             stderr="podman image ls logs",
         )
-        with pytest.raises(NotAvailableContainerTechException):
+        with pytest.raises(errors.NotAvailableContainerTechException):
             provider.is_available()
 
     def test_is_available_works(self, provider: Container, fp: FakeProcess) -> None:
@@ -45,7 +41,7 @@ class TestContainer(IsolationProviderTest):
         No exception should be raised when the "podman image ls" can return properly.
         """
         fp.register_subprocess(
-            [provider.get_runtime(), "image", "ls"],
+            [container_utils.get_runtime(), "image", "ls"],
         )
         provider.is_available()
 
@@ -55,13 +51,13 @@ class TestContainer(IsolationProviderTest):
         """When an image installation fails, an exception should be raised"""
 
         fp.register_subprocess(
-            [provider.get_runtime(), "image", "ls"],
+            [container_utils.get_runtime(), "image", "ls"],
         )
 
         # First check should return nothing.
         fp.register_subprocess(
             [
-                provider.get_runtime(),
+                container_utils.get_runtime(),
                 "image",
                 "list",
                 "--format",
@@ -75,11 +71,11 @@ class TestContainer(IsolationProviderTest):
         mocker.patch("gzip.open", mocker.mock_open(read_data=""))
 
         fp.register_subprocess(
-            [provider.get_runtime(), "load"],
+            [container_utils.get_runtime(), "load"],
             returncode=-1,
         )
 
-        with pytest.raises(ImageInstallationException):
+        with pytest.raises(errors.ImageInstallationException):
             provider.install()
 
     def test_install_raises_if_still_not_installed(
@@ -88,13 +84,13 @@ class TestContainer(IsolationProviderTest):
         """When an image keep being not installed, it should return False"""
 
         fp.register_subprocess(
-            [provider.get_runtime(), "image", "ls"],
+            [container_utils.get_runtime(), "image", "ls"],
         )
 
         # First check should return nothing.
         fp.register_subprocess(
             [
-                provider.get_runtime(),
+                container_utils.get_runtime(),
                 "image",
                 "list",
                 "--format",
@@ -107,9 +103,9 @@ class TestContainer(IsolationProviderTest):
         # Patch gzip.open and podman load so that it works
         mocker.patch("gzip.open", mocker.mock_open(read_data=""))
         fp.register_subprocess(
-            [provider.get_runtime(), "load"],
+            [container_utils.get_runtime(), "load"],
         )
-        with pytest.raises(ImageNotPresentException):
+        with pytest.raises(errors.ImageNotPresentException):
             provider.install()