Reorganize the registry.py module to be simpler

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/almet/dangerzone/dangerzone"
+CONTAINER_NAME = "ghcr.io/freedomofpress/dangerzone/dangerzone"
 
 log = logging.getLogger(__name__)
 
@@ -111,7 +111,7 @@ def delete_image_tag(tag: str) -> None:
         )
 
 
-def load_image_tarball_in_memory() -> None:
+def load_image_tarball_from_gzip() -> None:
     log.info("Installing Dangerzone container image...")
     p = subprocess.Popen(
         [get_runtime(), "load"],
@@ -142,7 +142,7 @@ def load_image_tarball_in_memory() -> None:
     log.info("Successfully installed container image from")
 
 
-def load_image_tarball_file(tarball_path: str) -> 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)
 

dangerzone/updater/attestations.py

@@ -3,7 +3,6 @@ from tempfile import NamedTemporaryFile
 
 from . import cosign
 
-
 # NOTE: You can grab the SLSA attestation for an image/tag pair with the following
 # commands:
 #
@@ -51,7 +50,11 @@ 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

@@ -6,9 +6,9 @@ import click
 
 from . import attestations, errors, log, registry, signatures
 
-DEFAULT_REPOSITORY = "almet/dangerzone"
+DEFAULT_REPOSITORY = "freedomofpress/dangerzone"
 DEFAULT_BRANCH = "main"
-DEFAULT_IMAGE_NAME = "ghcr.io/almet/dangerzone/dangerzone"
+DEFAULT_IMAGE_NAME = "ghcr.io/freedomofpress/dangerzone/dangerzone"
 
 
 @click.group()
@@ -97,8 +97,8 @@ def list_remote_tags(image: str) -> None:
 @main.command()
 @click.argument("image")
 def get_manifest(image: str) -> None:
-    """Retrieves a remove manifest for a given image and displays it."""
-    click.echo(registry.get_manifest(image))
+    """Retrieves a remote manifest for a given image and displays it."""
+    click.echo(registry.get_manifest(image).content)
 
 
 @main.command()

dangerzone/updater/registry.py

@@ -28,13 +28,7 @@ ACCEPT_MANIFESTS_HEADER = ",".join(
 )
 
 
-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}"
+Image = namedtuple("Image", ["registry", "namespace", "image_name", "tag"])
 
 
 def parse_image_location(input_string: str) -> Image:
@@ -58,102 +52,67 @@ def parse_image_location(input_string: str) -> Image:
     )
 
 
-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 _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}"}
 
 
-# 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 _url(image):
+    return f"https://{image.registry}/v2/{image.namespace}/{image.image_name}"
 
 
 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)
-    client = RegistryClient(image)
-    resp = client.get_manifest(image.tag)
-    return resp.content
+    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()

dangerzone/updater/signatures.py

@@ -152,34 +152,6 @@ def write_log_index(log_index: int) -> None:
         f.write(str(log_index))
 
 
-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
-
-
 def _get_blob(tmpdir: str, digest: str) -> Path:
     return Path(tmpdir) / "blobs" / "sha256" / digest.replace("sha256:", "")
 
@@ -252,7 +224,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_file(temporary_tar.name)
+            runtime.load_image_tarball_from_tar(temporary_tar.name)
             runtime.tag_image_by_digest(image_digest, image_name)
 
     store_signatures(signatures, image_digest, pubkey)
@@ -431,3 +403,31 @@ 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