Check signatures before invoking the container.

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.
This commit is contained in:
Alexis Métaireau 2025-02-05 16:54:13 +01:00
parent 60c144aab0
commit c2d37dfb04
No known key found for this signature in database
GPG key ID: C65C7A89A8FFC56E
5 changed files with 42 additions and 49 deletions

View file

@ -8,7 +8,8 @@ from typing import List, Optional, Tuple
from . import errors from . import errors
from .util import get_resource_path, get_subprocess_startupinfo from .util import get_resource_path, get_subprocess_startupinfo
CONTAINER_NAME = "dangerzone.rocks/dangerzone" OLD_CONTAINER_NAME = "dangerzone.rocks/dangerzone"
CONTAINER_NAME = "ghcr.io/almet/dangerzone/dangerzone"
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@ -110,12 +111,6 @@ def delete_image_tag(tag: str) -> None:
) )
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_in_memory() -> None: def load_image_tarball_in_memory() -> None:
log.info("Installing Dangerzone container image...") log.info("Installing Dangerzone container image...")
p = subprocess.Popen( p = subprocess.Popen(

View file

@ -5,7 +5,7 @@ import shlex
import subprocess import subprocess
from typing import List, Tuple from typing import List, Tuple
from .. import container_utils, errors from .. import container_utils, errors, updater
from ..document import Document from ..document import Document
from ..util import get_resource_path, get_subprocess_startupinfo from ..util import get_resource_path, get_subprocess_startupinfo
from .base import IsolationProvider, terminate_process_group from .base import IsolationProvider, terminate_process_group
@ -78,40 +78,22 @@ class Container(IsolationProvider):
@staticmethod @staticmethod
def install() -> bool: def install() -> bool:
"""Install the container image tarball, or verify that it's already installed. """Check if an update is available and install it if necessary."""
# XXX Do this only if users have optted in to auto-updates
Perform the following actions: # # Load the image tarball into the container runtime.
1. Get the tags of any locally available images that match Dangerzone's image update_available, image_digest = updater.is_update_available(
name. container_utils.CONTAINER_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. if update_available:
- Else, prune the older container images and continue. updater.upgrade_container_image(
3. Load the image tarball and make sure it matches the expected tag. container_utils.CONTAINER_NAME,
""" image_digest,
old_tags = container_utils.list_image_tags() updater.DEFAULT_PUBKEY_LOCATION,
expected_tag = container_utils.get_expected_tag()
if expected_tag not in old_tags:
# Prune older container images.
log.info(
f"Could not find a Dangerzone container image with tag '{expected_tag}'"
) )
for tag in old_tags:
container_utils.delete_image_tag(tag)
else:
return True
# Load the image tarball into the container runtime. updater.verify_local_image(
container_utils.load_image_tarball_in_memory() container_utils.CONTAINER_NAME, updater.DEFAULT_PUBKEY_LOCATION
# 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_utils.list_image_tags()
if expected_tag not in new_tags:
raise errors.ImageNotPresentException(
f"Could not find expected tag '{expected_tag}' after loading the"
" container image tarball"
) )
return True return True
@ -193,6 +175,13 @@ class Container(IsolationProvider):
name: str, name: str,
) -> subprocess.Popen: ) -> subprocess.Popen:
container_runtime = container_utils.get_runtime() container_runtime = container_utils.get_runtime()
image_digest = container_utils.get_local_image_digest(
container_utils.CONTAINER_NAME
)
updater.verify_local_image(
container_utils.CONTAINER_NAME, updater.DEFAULT_PUBKEY_LOCATION
)
security_args = self.get_runtime_security_args() security_args = self.get_runtime_security_args()
debug_args = [] debug_args = []
if self.debug: if self.debug:
@ -201,9 +190,7 @@ class Container(IsolationProvider):
enable_stdin = ["-i"] enable_stdin = ["-i"]
set_name = ["--name", name] set_name = ["--name", name]
prevent_leakage_args = ["--rm"] prevent_leakage_args = ["--rm"]
image_name = [ image_name = [container_utils.CONTAINER_NAME + "@sha256:" + image_digest]
container_utils.CONTAINER_NAME + ":" + container_utils.get_expected_tag()
]
args = ( args = (
["run"] ["run"]
+ security_args + security_args

View file

@ -1,3 +1,10 @@
import logging import logging
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
from .signatures import (
DEFAULT_PUBKEY_LOCATION,
is_update_available,
upgrade_container_image,
verify_local_image,
)

View file

@ -4,13 +4,11 @@ import logging
import click import click
from ..util import get_resource_path
from . import attestations, errors, log, registry, signatures from . import attestations, errors, log, registry, signatures
DEFAULT_REPOSITORY = "freedomofpress/dangerzone" DEFAULT_REPOSITORY = "freedomofpress/dangerzone"
DEFAULT_BRANCH = "main" DEFAULT_BRANCH = "main"
DEFAULT_IMAGE_NAME = "ghcr.io/freedomofpress/dangerzone/dangerzone" DEFAULT_IMAGE_NAME = "ghcr.io/freedomofpress/dangerzone/dangerzone"
PUBKEY_DEFAULT_LOCATION = get_resource_path("freedomofpress-dangerzone-pub.key")
@click.group() @click.group()
@ -26,7 +24,7 @@ def main(debug: bool) -> None:
@main.command() @main.command()
@click.argument("image", default=DEFAULT_IMAGE_NAME) @click.argument("image", default=DEFAULT_IMAGE_NAME)
@click.option("--pubkey", default=PUBKEY_DEFAULT_LOCATION) @click.option("--pubkey", default=signatures.DEFAULT_PUBKEY_LOCATION)
def upgrade(image: str, pubkey: str) -> None: def upgrade(image: str, pubkey: str) -> None:
"""Upgrade the image to the latest signed version.""" """Upgrade the image to the latest signed version."""
manifest_digest = registry.get_manifest_digest(image) manifest_digest = registry.get_manifest_digest(image)

View file

@ -11,6 +11,7 @@ from tempfile import NamedTemporaryFile, TemporaryDirectory
from typing import Dict, List, Optional, Tuple from typing import Dict, List, Optional, Tuple
from .. import container_utils as runtime from .. import container_utils as runtime
from ..util import get_resource_path
from . import cosign, errors, log, registry from . import cosign, errors, log, registry
try: try:
@ -24,6 +25,7 @@ def get_config_dir() -> Path:
# XXX Store this somewhere else. # XXX Store this somewhere else.
DEFAULT_PUBKEY_LOCATION = get_resource_path("freedomofpress-dangerzone-pub.key")
SIGNATURES_PATH = get_config_dir() / "signatures" SIGNATURES_PATH = get_config_dir() / "signatures"
__all__ = [ __all__ = [
"verify_signature", "verify_signature",
@ -103,12 +105,15 @@ def verify_signature(signature: dict, image_digest: str, pubkey: str) -> bool:
return False return False
def is_update_available(image: str) -> bool: def is_update_available(image: str) -> (bool, Optional[str]):
remote_digest = registry.get_manifest_digest(image) remote_digest = registry.get_manifest_digest(image)
local_digest = runtime.get_local_image_digest(image) local_digest = runtime.get_local_image_digest(image)
log.debug("Remote digest: %s", remote_digest) log.debug("Remote digest: %s", remote_digest)
log.debug("Local digest: %s", local_digest) log.debug("Local digest: %s", local_digest)
return remote_digest != local_digest has_update = remote_digest != local_digest
if has_update:
return True, remote_digest
return False, None
def verify_signatures( def verify_signatures(
@ -124,7 +129,8 @@ def verify_signatures(
def upgrade_container_image(image: str, manifest_digest: str, pubkey: str) -> bool: def upgrade_container_image(image: str, manifest_digest: str, pubkey: str) -> bool:
"""Verify and upgrade the image to the latest, if signed.""" """Verify and upgrade the image to the latest, if signed."""
if not is_update_available(image): update_available, _ = is_update_available(image)
if not update_available:
raise errors.ImageAlreadyUpToDate("The image is already up to date") raise errors.ImageAlreadyUpToDate("The image is already up to date")
signatures = get_remote_signatures(image, manifest_digest) signatures = get_remote_signatures(image, manifest_digest)