Provide an is_update_available function

This function does all the needed checks before returning `True`, making it a good external API.

Under the hood, the registry now has an `is_new_remote_image_available`
which is just for checking the presence of a new image, but doesn't do
any verirications on it, and there is also a new `check_signatures_and_logindex` that ensures that these two are valid.

dangerzone/container_utils.py

@@ -14,6 +14,11 @@ CONTAINER_NAME = "ghcr.io/freedomofpress/dangerzone/dangerzone"
 log = logging.getLogger(__name__)
 
 
+def subprocess_run(*args, **kwargs) -> subprocess.CompletedProcess:
+    """subprocess.run with the correct startupinfo for Windows."""
+    return subprocess.run(*args, startupinfo=get_subprocess_startupinfo(), **kwargs)
+
+
 def get_runtime_name() -> str:
     if platform.system() == "Linux":
         return "podman"
@@ -39,9 +44,8 @@ def get_runtime_version() -> Tuple[int, int]:
 
     cmd = [runtime, "version", "-f", query]
     try:
-        version = subprocess.run(
+        version = subprocess_run(
             cmd,
-            startupinfo=get_subprocess_startupinfo(),
             capture_output=True,
             check=True,
         ).stdout.decode()
@@ -144,8 +148,7 @@ def load_image_tarball_from_gzip() -> None:
 
 def load_image_tarball_from_tar(tarball_path: str) -> None:
     cmd = [get_runtime(), "load", "-i", tarball_path]
-    subprocess.run(cmd, startupinfo=get_subprocess_startupinfo(), check=True)
-
+    subprocess_run(cmd, check=True)
     log.info("Successfully installed container image from %s", tarball_path)
 
 
@@ -156,7 +159,7 @@ def tag_image_by_digest(digest: str, tag: str) -> None:
     image_id = get_image_id_by_digest(digest)
     cmd = [get_runtime(), "tag", image_id, tag]
     log.debug(" ".join(cmd))
-    subprocess.run(cmd, startupinfo=get_subprocess_startupinfo(), check=True)
+    subprocess_run(cmd, check=True)
 
 
 def get_image_id_by_digest(digest: str) -> str:
@@ -172,31 +175,37 @@ def get_image_id_by_digest(digest: str) -> str:
         "{{.Id}}",
     ]
     log.debug(" ".join(cmd))
-    process = subprocess.run(
-        cmd, startupinfo=get_subprocess_startupinfo(), check=True, capture_output=True
-    )
+    process = subprocess_run(cmd, check=True, capture_output=True)
     # In case we have multiple lines, we only want the first one.
     return process.stdout.decode().strip().split("\n")[0]
 
 
-def container_pull(image: str) -> bool:
+def container_pull(image: str, manifest_digest: str):
     """Pull a container image from a registry."""
-    cmd = [get_runtime_name(), "pull", f"{image}"]
-    process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
-    process.communicate()
-    return process.returncode == 0
+    cmd = [get_runtime_name(), "pull", f"{image}@sha256:{manifest_digest}"]
+    try:
+        subprocess_run(cmd, check=True)
+    except subprocess.CalledProcessError as e:
+        raise errors.ContainerPullException(
+            f"Could not pull the container image: {e}"
+        ) from e
 
 
 def get_local_image_digest(image: str) -> str:
     """
     Returns a image hash from a local image name
     """
-    # Get the image hash from the podman images command, as
-    # podman inspect returns a the digest of the architecture-bound image
+    # Get the image hash from the "podman images" command.
+    # It's not possible to use "podman inspect" here as it
+    # returns the digest of the architecture-bound image
     cmd = [get_runtime_name(), "images", image, "--format", "{{.Digest}}"]
     log.debug(" ".join(cmd))
     try:
-        result = subprocess.run(cmd, capture_output=True, check=True)
+        result = subprocess_run(
+            cmd,
+            capture_output=True,
+            check=True,
+        )
         lines = result.stdout.decode().strip().split("\n")
         if len(lines) != 1:
             raise errors.MultipleImagesFoundException(

dangerzone/errors.py

@@ -122,25 +122,33 @@ def handle_document_errors(func: F) -> F:
 #### Container-related errors
 
 
-class ImageNotPresentException(Exception):
+class ContainerException(Exception):
     pass
 
 
-class MultipleImagesFoundException(Exception):
+class ImageNotPresentException(ContainerException):
     pass
 
 
-class ImageInstallationException(Exception):
+class MultipleImagesFoundException(ContainerException):
     pass
 
 
-class NoContainerTechException(Exception):
+class ImageInstallationException(ContainerException):
+    pass
+
+
+class NoContainerTechException(ContainerException):
     def __init__(self, container_tech: str) -> None:
         super().__init__(f"{container_tech} is not installed")
 
 
-class NotAvailableContainerTechException(Exception):
+class NotAvailableContainerTechException(ContainerException):
     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 ContainerPullException(ContainerException):
+    pass

dangerzone/updater/cli.py

@@ -29,15 +29,17 @@ def upgrade(image: str, pubkey: str) -> None:
     """Upgrade the image to the latest signed version."""
     manifest_digest = registry.get_manifest_digest(image)
     try:
-        is_upgraded = signatures.upgrade_container_image(image, manifest_digest, pubkey)
-        if is_upgraded:
-            click.echo(f"✅ The local image {image} has been upgraded")
-            click.echo(f"✅ The image has been signed with {pubkey}")
-            click.echo(f"✅ Signatures has been verified and stored locally")
+        signatures.upgrade_container_image(image, manifest_digest, pubkey)
+        click.echo(f"✅ The local image {image} has been upgraded")
+        click.echo(f"✅ The image has been signed with {pubkey}")
+        click.echo(f"✅ Signatures has been verified and stored locally")
 
     except errors.ImageAlreadyUpToDate as e:
         click.echo(f"✅ {e}")
         raise click.Abort()
+    except Exception as e:
+        click.echo(f"❌ {e}")
+        raise click.Abort()
 
 
 @main.command()

dangerzone/updater/registry.py

@@ -5,6 +5,8 @@ from typing import Dict, Optional, Tuple
 
 import requests
 
+from .. import container_utils as runtime
+from .. import errors as dzerrors
 from . import errors, log
 
 __all__ = [
@@ -114,3 +116,24 @@ def get_manifest_digest(
         tag_manifest_content = get_manifest(image_str).content
 
     return sha256(tag_manifest_content).hexdigest()
+
+
+def is_new_remote_image_available(image_str: str) -> Tuple[bool, str]:
+    """
+    Check if a new remote image is available on the registry.
+    """
+    remote_digest = get_manifest_digest(image_str)
+    image = parse_image_location(image_str)
+    if image.digest:
+        local_digest = image.digest
+    else:
+        try:
+            local_digest = runtime.get_local_image_digest(image_str)
+        except dzerrors.ImageNotPresentException:
+            log.debug("No local image found")
+            return True, remote_digest
+
+    log.debug("Remote digest: %s", remote_digest)
+    log.debug("Local digest: %s", local_digest)
+
+    return (remote_digest != local_digest, remote_digest)

dangerzone/updater/signatures.py

@@ -22,13 +22,17 @@ except ImportError:
     import appdirs as platformdirs  # type: ignore[no-redef]
 
 
-def get_config_dir() -> Path:
-    return Path(platformdirs.user_config_dir("dangerzone"))
+def appdata_dir() -> Path:
+    return Path(platformdirs.user_data_dir("dangerzone"))
 
 
+# RELEASE: Bump this value to the log index of the latest signature
+# to ensures the software can't upgrade to container images that predates it.
+DEFAULT_LOG_INDEX = 0
+
 # XXX Store this somewhere else.
 DEFAULT_PUBKEY_LOCATION = get_resource_path("freedomofpress-dangerzone-pub.key")
-SIGNATURES_PATH = get_config_dir() / "signatures"
+SIGNATURES_PATH = appdata_dir() / "signatures"
 LAST_LOG_INDEX = SIGNATURES_PATH / "last_log_index"
 
 __all__ = [
@@ -61,9 +65,14 @@ def signature_to_bundle(sig: Dict) -> Dict:
     }
 
 
-def verify_signature(signature: dict, image_digest: str, pubkey: str | Path) -> bool:
-    """Verify a signature against a given public key"""
-    # XXX - Also verfy the identity/docker-reference field against the expected value
+def verify_signature(signature: dict, image_digest: str, pubkey: str | Path) -> None:
+    """
+    Verifies that:
+
+    - the signature has been signed by the given public key
+    - the signature matches the given image digest
+    """
+    # XXX - Also verify the identity/docker-reference field against the expected value
     # e.g. ghcr.io/freedomofpress/dangerzone/dangerzone
 
     cosign.ensure_installed()
@@ -79,7 +88,8 @@ def verify_signature(signature: dict, image_digest: str, pubkey: str | Path) ->
         )
     if payload_digest != f"sha256:{image_digest}":
         raise errors.SignatureMismatch(
-            f"The signature does not match the image digest ({payload_digest}, {image_digest})"
+            "The given signature does not match the expected image digest "
+            f"({payload_digest}, {image_digest})"
         )
 
     with (
@@ -106,14 +116,10 @@ def verify_signature(signature: dict, image_digest: str, pubkey: str | Path) ->
         ]
         log.debug(" ".join(cmd))
         result = subprocess.run(cmd, capture_output=True)
-        if result.returncode != 0:
-            # XXX Raise instead?
+        if result.returncode != 0 or result.stderr != b"Verified OK\n":
             log.debug("Failed to verify signature", result.stderr)
             raise errors.SignatureVerificationError("Failed to verify signature")
-        if result.stderr == b"Verified OK\n":
-            log.debug("Signature verified")
-            return True
-    return False
+        log.debug("Signature verified")
 
 
 class Signature:
@@ -130,19 +136,39 @@ class Signature:
         return full_digest.replace("sha256:", "")
 
 
-def is_update_available(image: str) -> Tuple[bool, Optional[str]]:
-    remote_digest = registry.get_manifest_digest(image)
+def is_update_available(image_str: str, pubkey: str) -> Tuple[bool, Optional[str]]:
+    """
+    Check if a new image is available, doing all the necessary checks ensuring it
+    would be safe to upgrade.
+    """
+    new_image_available, remote_digest = registry.is_new_remote_image_available(
+        image_str
+    )
+    if not new_image_available:
+        return False, None
+
     try:
-        local_digest = runtime.get_local_image_digest(image)
-    except dzerrors.ImageNotPresentException:
-        log.debug("No local image found")
+        check_signatures_and_logindex(image_str, remote_digest, pubkey)
         return True, remote_digest
-    log.debug("Remote digest: %s", remote_digest)
-    log.debug("Local digest: %s", local_digest)
-    has_update = remote_digest != local_digest
-    if has_update:
-        return True, remote_digest
-    return False, None
+    except errors.InvalidLogIndex:
+        return False, None
+
+
+def check_signatures_and_logindex(
+    image_str: str, remote_digest: str, pubkey: str
+) -> list[Dict]:
+    signatures = get_remote_signatures(image_str, remote_digest)
+    verify_signatures(signatures, remote_digest, pubkey)
+
+    incoming_log_index = get_log_index_from_signatures(signatures)
+    last_log_index = get_last_log_index()
+
+    if incoming_log_index < last_log_index:
+        raise errors.InvalidLogIndex(
+            f"The incoming log index ({incoming_log_index}) is "
+            f"lower than the last known log index ({last_log_index})"
+        )
+    return signatures
 
 
 def verify_signatures(
@@ -154,17 +180,14 @@ def verify_signatures(
         raise errors.SignatureVerificationError("No signatures found")
 
     for signature in signatures:
-        if not verify_signature(signature, image_digest, pubkey):
-            msg = f"Unable to verify signature for {image_digest} with pubkey {pubkey}"
-            raise errors.SignatureVerificationError(msg)
-
+        verify_signature(signature, image_digest, pubkey)
     return True
 
 
 def get_last_log_index() -> int:
     SIGNATURES_PATH.mkdir(parents=True, exist_ok=True)
     if not LAST_LOG_INDEX.exists():
-        return 0
+        return DEFAULT_LOG_INDEX
 
     with open(LAST_LOG_INDEX) as f:
         return int(f.read())
@@ -364,6 +387,8 @@ def store_signatures(signatures: list[Dict], image_digest: str, pubkey: str) ->
 
     It can be converted to the one expected by cosign verify --bundle with
     the `signature_to_bundle()` function.
+
+    This function must be used only if the provided signatures have been verified.
     """
 
     def _get_digest(sig: Dict) -> str:
@@ -453,29 +478,15 @@ def prepare_airgapped_archive(image_name: str, destination: str) -> None:
             archive.add(tmpdir, arcname=".")
 
 
-def upgrade_container_image(image: str, manifest_digest: str, pubkey: str) -> bool:
+def upgrade_container_image(image: str, manifest_digest: str, pubkey: str) -> str:
     """Verify and upgrade the image to the latest, if signed."""
-    update_available, _ = is_update_available(image)
+    update_available, remote_digest = registry.is_new_remote_image_available(image)
     if not update_available:
         raise errors.ImageAlreadyUpToDate("The image is already up to date")
 
-    signatures = get_remote_signatures(image, manifest_digest)
-    verify_signatures(signatures, manifest_digest, pubkey)
+    signatures = check_signatures_and_logindex(image, remote_digest, pubkey)
+    runtime.container_pull(image, manifest_digest)
 
-    # Ensure that we only upgrade if the log index is higher than the last known one
-    incoming_log_index = get_log_index_from_signatures(signatures)
-    last_log_index = get_last_log_index()
-
-    if incoming_log_index < last_log_index:
-        raise errors.InvalidLogIndex(
-            "The log index is not higher than the last known one"
-        )
-
-    # let's upgrade the image
-    # XXX Use the image digest here to avoid race conditions
-    upgraded = runtime.container_pull(image)
-
-    # At this point, the signatures are verified
-    # We store the signatures just now to avoid storing unverified signatures
+    # Store the signatures just now to avoid storing them unverified
     store_signatures(signatures, manifest_digest, pubkey)
-    return upgraded
+    return manifest_digest

tests/conftest.py

@@ -13,12 +13,6 @@ from dangerzone.gui import Application
 sys.dangerzone_dev = True  # type: ignore[attr-defined]
 
 
-ASSETS_PATH = Path(__file__).parent / "assets"
-TEST_PUBKEY_PATH = ASSETS_PATH / "test.pub.key"
-INVALID_SIGNATURES_PATH = ASSETS_PATH / "signatures" / "invalid"
-VALID_SIGNATURES_PATH = ASSETS_PATH / "signatures" / "valid"
-TEMPERED_SIGNATURES_PATH = ASSETS_PATH / "signatures" / "tempered"
-
 
 # Use this fixture to make `pytest-qt` invoke our custom QApplication.
 # See https://pytest-qt.readthedocs.io/en/latest/qapplication.html#testing-custom-qapplications
@@ -140,10 +134,6 @@ for_each_doc = pytest.mark.parametrize(
 )
 
 
-@pytest.fixture
-def signature():
-    return {}
-
 
 # External Docs - base64 docs encoded for externally sourced documents
 # XXX to reduce the chance of accidentally opening them

tests/test_signatures.py

@@ -9,7 +9,6 @@ from dangerzone import errors as dzerrors
 from dangerzone.updater import errors
 from dangerzone.updater.signatures import (
     Signature,
-    get_config_dir,
     get_last_log_index,
     get_log_index_from_signatures,
     get_remote_signatures,