Fix cli.py

Also, check for new container images when starting the application.
This replaces the usage of `share/image-id.txt` to ensure the image is trusted.

.grype.yaml

@@ -37,12 +37,3 @@ ignore:
   #       [bookworm] - raptor2 <postponed> (Minor issue, revisit when fixed upstream)
   #
   - vulnerability: CVE-2024-57823
-  # CVE-2025-0665
-  # ==============
-  #
-  # Debian tracker: https://security-tracker.debian.org/tracker/CVE-2025-0665
-  # Verdict: Dangerzone is not affected because the vulnerable code is not
-  # present in Debian Bookworm. Also, libcurl is an HTTP client, and the
-  # Dangerzone container does not make any network calls.
-  - vulnerability: CVE-2025-0665
-

dangerzone/container_utils.py

@@ -9,7 +9,7 @@ from . import errors
 from .util import get_resource_path, get_subprocess_startupinfo
 
 OLD_CONTAINER_NAME = "dangerzone.rocks/dangerzone"
-CONTAINER_NAME = "ghcr.io/freedomofpress/dangerzone/dangerzone"
+CONTAINER_NAME = "ghcr.io/almet/dangerzone/dangerzone"
 
 log = logging.getLogger(__name__)
 
@@ -111,7 +111,7 @@ def delete_image_tag(tag: str) -> None:
         )
 
 
-def load_image_tarball_from_gzip() -> None:
+def load_image_tarball_in_memory() -> None:
     log.info("Installing Dangerzone container image...")
     p = subprocess.Popen(
         [get_runtime(), "load"],
@@ -142,7 +142,7 @@ def load_image_tarball_from_gzip() -> None:
     log.info("Successfully installed container image from")
 
 
-def load_image_tarball_from_tar(tarball_path: str) -> None:
+def load_image_tarball_file(tarball_path: str) -> None:
     cmd = [get_runtime(), "load", "-i", tarball_path]
     subprocess.run(cmd, startupinfo=get_subprocess_startupinfo(), check=True)
 

dangerzone/updater/attestations.py

@@ -3,6 +3,7 @@ from tempfile import NamedTemporaryFile
 
 from . import cosign
 
+
 # NOTE: You can grab the SLSA attestation for an image/tag pair with the following
 # commands:
 #
@@ -50,11 +51,7 @@ def generate_cue_policy(repo, workflow, commit, branch):
 
 
 def verify(
-    image_name: str,
-    branch: str,
-    commit: str,
-    repository: str,
-    workflow: str,
+    image_name: str, branch: str, commit: str, repository: str, workflow: str,
 ) -> bool:
     """
     Look up the image attestation to see if the image has been built

dangerzone/updater/cli.py

@@ -97,8 +97,8 @@ def list_remote_tags(image: str) -> None:
 @main.command()
 @click.argument("image")
 def get_manifest(image: str) -> None:
-    """Retrieves a remote manifest for a given image and displays it."""
-    click.echo(registry.get_manifest(image).content)
+    """Retrieves a remove manifest for a given image and displays it."""
+    click.echo(registry.get_manifest(image))
 
 
 @main.command()
@@ -121,7 +121,7 @@ def get_manifest(image: str) -> None:
 )
 @click.option(
     "--workflow",
-    default=".github/workflows/release-container-image.yml",
+    default=".github/workflows/multi_arch_build.yml",
     help="The path of the GitHub actions workflow this image was created from",
 )
 def attest_provenance(

dangerzone/updater/errors.py

@@ -52,7 +52,3 @@ class LocalSignatureNotFound(SignatureError):
 
 class CosignNotInstalledError(SignatureError):
     pass
-
-
-class InvalidLogIndex(SignatureError):
-    pass

dangerzone/updater/registry.py

@@ -28,7 +28,13 @@ ACCEPT_MANIFESTS_HEADER = ",".join(
 )
 
 
-Image = namedtuple("Image", ["registry", "namespace", "image_name", "tag"])
+class Image(namedtuple("Image", ["registry", "namespace", "image_name", "tag"])):
+    __slots__ = ()
+
+    @property
+    def full_name(self) -> str:
+        tag = f":{self.tag}" if self.tag else ""
+        return f"{self.registry}/{self.namespace}/{self.image_name}{tag}"
 
 
 def parse_image_location(input_string: str) -> Image:
@@ -52,67 +58,102 @@ def parse_image_location(input_string: str) -> Image:
     )
 
 
-def _get_auth_header(image) -> Dict[str, str]:
-    auth_url = f"https://{image.registry}/token"
-    response = requests.get(
-        auth_url,
-        params={
-            "service": f"{image.registry}",
-            "scope": f"repository:{image.namespace}/{image.image_name}:pull",
-        },
-    )
-    response.raise_for_status()
-    token = response.json()["token"]
-    return {"Authorization": f"Bearer {token}"}
+class RegistryClient:
+    def __init__(
+        self,
+        image: Image | str,
+    ):
+        if isinstance(image, str):
+            image = parse_image_location(image)
+
+        self._image = image
+        self._registry = image.registry
+        self._namespace = image.namespace
+        self._image_name = image.image_name
+        self._auth_token = None
+        self._base_url = f"https://{self._registry}"
+        self._image_url = f"{self._base_url}/v2/{self._namespace}/{self._image_name}"
+
+    def get_auth_token(self) -> Optional[str]:
+        if not self._auth_token:
+            auth_url = f"{self._base_url}/token"
+            response = requests.get(
+                auth_url,
+                params={
+                    "service": f"{self._registry}",
+                    "scope": f"repository:{self._namespace}/{self._image_name}:pull",
+                },
+            )
+            response.raise_for_status()
+            self._auth_token = response.json()["token"]
+        return self._auth_token
+
+    def get_auth_header(self) -> Dict[str, str]:
+        return {"Authorization": f"Bearer {self.get_auth_token()}"}
+
+    def list_tags(self) -> list:
+        url = f"{self._image_url}/tags/list"
+        response = requests.get(url, headers=self.get_auth_header())
+        response.raise_for_status()
+        tags = response.json().get("tags", [])
+        return tags
+
+    def get_manifest(
+        self,
+        tag: str,
+    ) -> requests.Response:
+        """Get manifest information for a specific tag"""
+        manifest_url = f"{self._image_url}/manifests/{tag}"
+        headers = {
+            "Accept": ACCEPT_MANIFESTS_HEADER,
+            "Authorization": f"Bearer {self.get_auth_token()}",
+        }
+
+        response = requests.get(manifest_url, headers=headers)
+        response.raise_for_status()
+        return response
+
+    def list_manifests(self, tag: str) -> list:
+        return (
+            self.get_manifest(
+                tag,
+            )
+            .json()
+            .get("manifests")
+        )
+
+    def get_blob(self, digest: str) -> requests.Response:
+        url = f"{self._image_url}/blobs/{digest}"
+        response = requests.get(
+            url,
+            headers={
+                "Authorization": f"Bearer {self.get_auth_token()}",
+            },
+        )
+        response.raise_for_status()
+        return response
+
+    def get_manifest_digest(
+        self, tag: str, tag_manifest_content: Optional[bytes] = None
+    ) -> str:
+        if not tag_manifest_content:
+            tag_manifest_content = self.get_manifest(tag).content
+
+        return sha256(tag_manifest_content).hexdigest()
 
 
-def _url(image):
-    return f"https://{image.registry}/v2/{image.namespace}/{image.image_name}"
+# XXX Refactor this with regular functions rather than a class
+def get_manifest_digest(image_str: str) -> str:
+    image = parse_image_location(image_str)
+    return RegistryClient(image).get_manifest_digest(image.tag)
 
 
 def list_tags(image_str: str) -> list:
+    return RegistryClient(image_str).list_tags()
+
+
+def get_manifest(image_str: str) -> bytes:
     image = parse_image_location(image_str)
-    url = f"{_url(image)}/tags/list"
-    response = requests.get(url, headers=_get_auth_header(image))
-    response.raise_for_status()
-    tags = response.json().get("tags", [])
-    return tags
-
-
-def get_manifest(image_str) -> requests.Response:
-    """Get manifest information for a specific tag"""
-    image = parse_image_location(image_str)
-    manifest_url = f"{_url(image)}/manifests/{image.tag}"
-    headers = {
-        "Accept": ACCEPT_MANIFESTS_HEADER,
-    }
-    headers.update(_get_auth_header(image))
-
-    response = requests.get(manifest_url, headers=headers)
-    response.raise_for_status()
-    return response
-
-
-def list_manifests(image_str) -> list:
-    return get_manifest(image_str).json().get("manifests")
-
-
-def get_blob(image, digest: str) -> requests.Response:
-    response = requests.get(
-        f"{_url(image)}/blobs/{digest}",
-        headers={
-            "Authorization": f"Bearer {_get_auth_token(image)}",
-        },
-    )
-    response.raise_for_status()
-    return response
-
-
-def get_manifest_digest(
-    image_str: str, tag_manifest_content: Optional[bytes] = None
-) -> str:
-    image = parse_image_location(image_str)
-    if not tag_manifest_content:
-        tag_manifest_content = get_manifest(image).content
-
-    return sha256(tag_manifest_content).hexdigest()
+    client = RegistryClient(image)
+    resp = client.get_manifest(image.tag)
+    return resp.content

dangerzone/updater/signatures.py

@@ -4,7 +4,6 @@ import re
 import subprocess
 import tarfile
 from base64 import b64decode, b64encode
-from functools import reduce
 from hashlib import sha256
 from io import BytesIO
 from pathlib import Path
@@ -28,8 +27,6 @@ def get_config_dir() -> Path:
 # XXX Store this somewhere else.
 DEFAULT_PUBKEY_LOCATION = get_resource_path("freedomofpress-dangerzone-pub.key")
 SIGNATURES_PATH = get_config_dir() / "signatures"
-LAST_LOG_INDEX = SIGNATURES_PATH / "last_log_index"
-
 __all__ = [
     "verify_signature",
     "load_signatures",
@@ -130,26 +127,22 @@ def verify_signatures(
     return True
 
 
-def get_last_log_index() -> int:
-    SIGNATURES_PATH.mkdir(parents=True, exist_ok=True)
-    if not LAST_LOG_INDEX.exists():
-        return 0
+def upgrade_container_image(image: str, manifest_digest: str, pubkey: str) -> bool:
+    """Verify and upgrade the image to the latest, if signed."""
+    update_available, _ = is_update_available(image)
+    if not update_available:
+        raise errors.ImageAlreadyUpToDate("The image is already up to date")
 
-    with open(LAST_LOG_INDEX) as f:
-        return int(f.read())
+    signatures = get_remote_signatures(image, manifest_digest)
+    verify_signatures(signatures, manifest_digest, pubkey)
 
+    # At this point, the signatures are verified
+    # We store the signatures just now to avoid storing unverified signatures
+    store_signatures(signatures, manifest_digest, pubkey)
 
-def get_log_index_from_signatures(signatures: List[Dict]) -> int:
-    return reduce(
-        lambda acc, sig: max(acc, sig["Bundle"]["Payload"]["logIndex"]), signatures, 0
-    )
-
-
-def write_log_index(log_index: int) -> None:
-    last_log_index_path = SIGNATURES_PATH / "last_log_index"
-
-    with open(log_index, "w") as f:
-        f.write(str(log_index))
+    # let's upgrade the image
+    # XXX Use the image digest here to avoid race conditions
+    return runtime.container_pull(image)
 
 
 def _get_blob(tmpdir: str, digest: str) -> Path:
@@ -185,7 +178,7 @@ def upgrade_container_image_airgapped(container_tar: str, pubkey: str) -> str:
         if not cosign.verify_local_image(tmpdir, pubkey):
             raise errors.SignatureVerificationError()
 
-        # Remove the signatures from the archive, otherwise podman is not able to load it
+        # Remove the signatures from the archive.
         with open(Path(tmpdir) / "index.json") as f:
             index_json = json.load(f)
 
@@ -202,15 +195,6 @@ def upgrade_container_image_airgapped(container_tar: str, pubkey: str) -> str:
             image_name, signatures = convert_oci_images_signatures(json.load(f), tmpdir)
         log.info(f"Found image name: {image_name}")
 
-        # 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"
-            )
-
         image_digest = index_json["manifests"][0].get("digest").replace("sha256:", "")
 
         # Write the new index.json to the temp folder
@@ -224,7 +208,7 @@ def upgrade_container_image_airgapped(container_tar: str, pubkey: str) -> str:
                 archive.add(Path(tmpdir) / "oci-layout", arcname="oci-layout")
                 archive.add(Path(tmpdir) / "blobs", arcname="blobs")
 
-            runtime.load_image_tarball_from_tar(temporary_tar.name)
+            runtime.load_image_tarball_file(temporary_tar.name)
             runtime.tag_image_by_digest(image_digest, image_name)
 
     store_signatures(signatures, image_digest, pubkey)
@@ -299,13 +283,9 @@ def store_signatures(signatures: list[Dict], image_digest: str, pubkey: str) ->
     Store signatures locally in the SIGNATURE_PATH folder, like this:
 
     ~/.config/dangerzone/signatures/
-    ├── <pubkey-digest>
-    │   ├── <image-digest>.json
-    │   ├── <image-digest>.json
-    └── last_log_index
-
-    The last_log_index file is used to keep track of the last log index
-    processed by the updater.
+    └── <pubkey-digest>
+        └── <image-digest>.json
+        └── <image-digest>.json
 
     The format used in the `.json` file is the one of `cosign download
     signature`, which differs from the "bundle" one used afterwards.
@@ -364,7 +344,6 @@ def get_remote_signatures(image: str, digest: str) -> List[Dict]:
     """Retrieve the signatures from the registry, via `cosign download`."""
     cosign.ensure_installed()
 
-    # XXX: try/catch here
     process = subprocess.run(
         ["cosign", "download", "signature", f"{image}@sha256:{digest}"],
         capture_output=True,
@@ -403,31 +382,3 @@ def prepare_airgapped_archive(image_name, destination):
 
         with tarfile.open(destination, "w") as archive:
             archive.add(tmpdir, arcname=".")
-
-
-def upgrade_container_image(image: str, manifest_digest: str, pubkey: str) -> bool:
-    """Verify and upgrade the image to the latest, if signed."""
-    update_available, _ = is_update_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)
-
-    # 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_signatures(signatures, manifest_digest, pubkey)
-    return upgraded

docs/developer/independent-container-updates.md

@@ -22,13 +22,13 @@ In case of sucess, it will report back:
 
 ```
 🎉 Successfully verified image
-'ghcr.io/freedomofpress/dangerzone/dangerzone:<tag>@sha256:<digest>'
+'ghcr.io/freedomofpress/dangerzone/dangerzone:20250129-0.8.0-149-gbf2f5ac@sha256:4da441235e84e93518778827a5c5745d532d7a4079886e1647924bee7ef1c14d'
 and its associated claims:
 - ✅ SLSA Level 3 provenance
-- ✅ GitHub repo: freedomofpress/dangerzone
-- ✅ GitHub actions workflow: <workflow>
-- ✅ Git branch: <branch>
-- ✅ Git commit: <commit>
+- ✅ GitHub repo: apyrgio/dangerzone
+- ✅ GitHub actions workflow: .github/workflows/multi_arch_build.yml
+- ✅ Git branch: test/multi-arch
+- ✅ Git commit: bf2f5accc24bd15a4f5c869a7f0b03b8fe48dfb6
 ```
 
 ## Sign and publish the remote image
@@ -37,11 +37,11 @@ Once the image has been reproduced locally, we can add a signature to the contai
 and update the `latest` tag to point to the proper hash.
 
 ```bash
-cosign sign --sk ghcr.io/freedomofpress/dangerzone/dangerzone:${TAG}@sha256:${DIGEST}
+cosign sign --sk ghcr.io/freedomofpress/dangerzone/dangerzone:20250129-0.8.0-149-gbf2f5ac@sha256:4da441235e84e93518778827a5c5745d532d7a4079886e1647924bee7ef1c14d
 
 # And mark bump latest
 crane auth login ghcr.io -u USERNAME --password $(cat pat_token)
-crane tag ghcr.io/freedomofpress/dangerzone/dangerzone@sha256:${DIGEST} latest
+crane tag ghcr.io/freedomofpress/dangerzone/dangerzone@sha256:4da441235e84e93518778827a5c5745d532d7a4079886e1647924bee7ef1c14d latest
 ```
 
 ## Install updates
@@ -49,7 +49,7 @@ crane tag ghcr.io/freedomofpress/dangerzone/dangerzone@sha256:${DIGEST} latest
 To check if a new container image has been released, and update your local installation with it, you can use the following commands:
 
 ```bash
-dangerzone-image upgrade ghcr.io/freedomofpress/dangerzone/dangerzone
+dangerzone-image upgrade ghcr.io/almet/dangerzone/dangerzone
 ```
 
 ## Verify locally
@@ -57,7 +57,7 @@ dangerzone-image upgrade ghcr.io/freedomofpress/dangerzone/dangerzone
 You can verify that the image you have locally matches the stored signatures, and that these have been signed with a trusted public key:
 
 ```bash
-dangerzone-image verify-local ghcr.io/freedomofpress/dangerzone/dangerzone
+dangerzone-image verify-local ghcr.io/almet/dangerzone/dangerzone
 ```
 
 ## Installing image updates to air-gapped environments
@@ -73,7 +73,7 @@ This archive will contain all the needed material to validate that the new conta
 On the machine on which you prepare the packages:
 
 ```bash
-dangerzone-image prepare-archive --output dz-fa94872.tar ghcr.io/freedomofpress/dangerzone/dangerzone@sha256:<digest>
+dangerzone-image prepare-archive --output dz-fa94872.tar ghcr.io/almet/dangerzone/dangerzone@sha256:fa948726aac29a6ac49f01ec8fbbac18522b35b2491fdf716236a0b3502a2ca7
 ```
 
 On the airgapped machine, copy the file and run the following command: