feat: support tags and digest simultaneously

With this change, Connaisseur now supports the use of tags and digests simultaneously. The signature is still validated based on the digest, but the human readable aspect of the tag isn't lost.

connaisseur/flask_application.py

@@ -6,6 +6,7 @@
 from flask import Flask, jsonify, request
 from prometheus_flask_exporter import PrometheusMetrics, NO_PREFIX
 
+import connaisseur.constants as const
 from connaisseur.admission_request import AdmissionRequest
 from connaisseur.alert import send_alerts
 from connaisseur.config import Config
@@ -190,5 +191,5 @@ async def __validate_image(type_index, image, admission_request):
     msg = f'successful verification of image "{original_image}"'
     logging.info(__create_logging_msg(msg, **logging_context))
     if trusted_digest:
-        image.set_digest(trusted_digest)
+        image.digest, image.digest_algo = trusted_digest, const.SHA256
         return admission_request.wl_object.get_json_patch(image, type_, index)

connaisseur/image.py

@@ -1,6 +1,7 @@
 import re
 from typing import Optional
 
+import connaisseur.constants as const
 from connaisseur.exceptions import InvalidImageFormatError
 
 
@@ -26,62 +27,71 @@ class Image:
     name: str
     tag: Optional[str]
     digest: Optional[str]
+    digest_algo: Optional[str]
 
-    def __init__(self, image: str):
-        separator = r"[-._:@+]|--"
-        alphanum = r"[A-Za-z0-9]+"
-        component = f"{alphanum}(?:(?:{separator}){alphanum})*"
-        ref = f"^{component}(?:/{component})*$"
+    def __init__(self, image: str):  # pylint: disable=too-many-locals
 
-        # e.g. :v1, :3.7-alpine, @sha256:3e7a89...
-        tag_re = r"(?:(?:@sha256:([a-f0-9]{64}))|(?:\:([\w.-]+)))"
-
-        match = re.search(ref, image)
-        if not match:
-            msg = "{image} is not a valid image reference."
-            raise InvalidImageFormatError(message=msg, image=image)
-
-        name_tag = image.split("/")[-1]
-        search = re.search(tag_re, name_tag)
-        self.digest, self.tag = search.groups() if search else (None, "latest")
-        self.name = name_tag.removesuffix(":" + str(self.tag)).removesuffix(
-            "@sha256:" + str(self.digest)
+        # implements https://github.com/distribution/distribution/blob/main/reference/regexp.go
+        digest_hex = r"[0-9a-fA-F]{32,}"
+        digest_algorithm_component = r"[A-Za-z][A-Za-z0-9]*"
+        digest_algorithm_separator = r"[+._-]"
+        digest_algorithm = (
+            rf"{digest_algorithm_component}(?:{digest_algorithm_separator}"
+            rf"{digest_algorithm_component})*"
         )
-
-        first_comp = image.removesuffix(name_tag).split("/")[0]
-        self.registry = (
-            first_comp
-            if re.search(r"[.:]", first_comp)
-            or first_comp == "localhost"
-            or any(ele.isupper() for ele in first_comp)
-            else "docker.io"
-        )
-        self.repository = (
-            image.removesuffix(name_tag).removeprefix(self.registry)
-        ).strip("/") or ("library" if self.registry == "docker.io" else "")
-
-        if (self.repository + self.name).lower() != self.repository + self.name:
+        digest = rf"{digest_algorithm}:{digest_hex}"
+        tag = r"[\w][\w.-]{0,127}"
+        separator = r"[_.]|__|[-]*"
+        alpha_numeric = r"[a-z0-9]+"
+        path_component = rf"{alpha_numeric}(?:{separator}{alpha_numeric})*"
+        port = r"[0-9]+"
+        domain_component = r"(?:[a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9-]*[a-zA-Z0-9])"
+        domain_name = rf"{domain_component}(?:\.{domain_component})*"
+        ipv6 = r"\[(?:[a-fA-F0-9:]+)\]"
+        host = rf"(?:{domain_name}|{ipv6})"
+        domain = rf"{host}(?::{port})?"
+        name = rf"(?:{domain}/)?{path_component}(?:/{path_component})*"
+        reference = rf"^(?P<name>{name})(?::(?P<tag>{tag}))?(?:@(?P<digest>{digest}))?$"
+
+        match = re.search(reference, image)
+        if (not match) or (len(match.group("name")) > 255):
             msg = "{image} is not a valid image reference."
             raise InvalidImageFormatError(message=msg, image=image)
 
-    def set_digest(self, digest):
-        """
-        Set the digest to the given `digest`.
-        """
-        self.digest = digest
-
-    def has_digest(self) -> bool:
-        """
-        Return `True` if the image has a digest, `False` otherwise.
-        """
-        return self.digest is not None
+        name, tag, digest = match.groups()
+        components = name.split("/")
+        self.name = components[-1]
+        self.digest_algo, self.digest = digest.split(":") if digest else (None, None)
+        self.tag = tag or ("latest" if not self.digest else None)
+
+        if self.digest_algo and self.digest_algo != const.SHA256:
+            raise InvalidImageFormatError(
+                message="A digest algorithm of {digest_algo} is not supported. Use sha256 instead.",
+                digest_algo=self.digest_algo,
+            )
+
+        registry_repo = components[:-1]
+        try:
+            registry = registry_repo[0]
+            self.registry = (
+                registry
+                if re.search(r"[.:]", registry)
+                or registry == "localhost"
+                or any(ele.isupper() for ele in registry)
+                else "docker.io"
+            )
+            self.repository = "/".join(registry_repo).removeprefix(
+                f"{self.registry}"
+            ).removeprefix("/") or ("library" if self.registry == "docker.io" else None)
+        except IndexError:
+            self.registry = "docker.io"
+            self.repository = "library"
 
     def __str__(self):
-        repo_reg = "".join(
-            f"{item}/" for item in [self.registry, self.repository] if item
-        )
-        tag = f":{self.tag}" if not self.digest else f"@sha256:{self.digest}"
-        return f"{repo_reg}{self.name}{tag}"
+        repo_reg = "/".join(item for item in [self.registry, self.repository] if item)
+        tag = f":{self.tag}" if self.tag else ""
+        digest = f"@{self.digest_algo}:{self.digest}" if self.digest else ""
+        return f"{repo_reg}/{self.name}{tag}{digest}"
 
     def __eq__(self, other):
         return str(self) == str(other)

connaisseur/validators/cosign/cosign_validator.py

@@ -7,6 +7,7 @@
 
 from concurrent.futures import ThreadPoolExecutor
 
+import connaisseur.constants as const
 from connaisseur.exceptions import (
     CosignError,
     CosignTimeout,
@@ -162,7 +163,7 @@ def __get_cosign_validated_digests(self, image: str, trust_root: dict):
                         digest = sig_data["critical"]["image"].get(
                             "docker-manifest-digest", ""
                         )
-                        if re.match(r"sha256:[0-9A-Fa-f]{64}", digest) is None:
+                        if re.match(rf"{const.SHA256}:[0-9A-Fa-f]{{64}}", digest) is None:
                             msg = "Digest '{digest}' does not match expected digest pattern."
                             raise InvalidFormatException(message=msg, digest=digest)
                     except Exception as err:
@@ -180,7 +181,7 @@ def __get_cosign_validated_digests(self, image: str, trust_root: dict):
                             trust_root=trust_root["name"],
                         ) from err
                     # remove prefix 'sha256'
-                    digests.append(digest.removeprefix("sha256:"))
+                    digests.append(digest.removeprefix(f"{const.SHA256}:"))
                 except json.JSONDecodeError:
                     logging.info("non-json signature data from Cosign: %s", sig)
                     pass

connaisseur/validators/notaryv1/key_store.py

@@ -1,5 +1,6 @@
 from connaisseur.exceptions import NotFoundException
 from connaisseur.trust_root import TrustRoot
+import connaisseur.constants as const
 
 
 class KeyStore:
@@ -68,7 +69,7 @@ def update(self, trust_data):
             self.hashes.setdefault(
                 role,
                 (
-                    hashes[role].get("hashes", {}).get("sha256"),
+                    hashes[role].get("hashes", {}).get(const.SHA256),
                     hashes[role].get("length", 0),
                 ),
             )

connaisseur/validators/notaryv1/notaryv1_validator.py

@@ -3,10 +3,12 @@
 import datetime as dt
 import logging
 
+import connaisseur.constants as const
 from connaisseur.exceptions import (
     AmbiguousDigestError,
     InsufficientTrustDataError,
     NotFoundException,
+    ValidationError,
 )
 from connaisseur.image import Image
 from connaisseur.trust_root import TrustRoot
@@ -47,15 +49,12 @@ async def validate(
             image, req_delegations, root_key
         )
 
-        # search for digests or tag, depending on given image
-        search_image_targets = (
-            NotaryV1Validator.__search_image_targets_for_digest
-            if image.has_digest()
-            else NotaryV1Validator.__search_image_targets_for_tag
-        )
-        # filter out the searched for digests, if present
+        # search for digests in given targets
         digests = list(
-            map(lambda x: search_image_targets(x, image), signed_image_targets)
+            map(
+                lambda x: NotaryV1Validator.__search_image_targets(x, image),
+                signed_image_targets,
+            )
         )
 
         # in case certain delegations are needed, `signed_image_targets` should only
@@ -224,29 +223,36 @@ async def __process_chain_of_trust(
         return image_targets
 
     @staticmethod
-    def __search_image_targets_for_digest(trust_data: dict, image: Image):
-        """
-        Search in the `trust_data` for a signed digest, given an `image` with
-        digest.
-        """
-        image_digest = base64.b64encode(bytes.fromhex(image.digest)).decode("utf-8")
-        if image_digest in {data["hashes"]["sha256"] for data in trust_data.values()}:
-            return image.digest
-
+    def __search_image_targets(trust_data: dict, image: Image):
+        if image.tag:
+            if image.tag not in trust_data:
+                return None
+
+            base64_digest = trust_data[image.tag]["hashes"][const.SHA256]
+            digest = base64.b64decode(base64_digest).hex()
+
+            # if both tag and digest are given
+            if image.digest:
+                # validate if the digest in the trust_data found by the tag,
+                # matches the digest requested by the image reference
+                if digest == image.digest:
+                    return digest
+                else:
+                    raise ValidationError(
+                        message="Image tag and digest do not match.",
+                        tag=image.tag,
+                        tag_digest=digest,
+                        digest=image.digest,
+                    )
+            # if only the tag is given
+            return digest
+        # if only the digest is given
+        elif image.digest:
+            digest = base64.b64encode(bytes.fromhex(image.digest)).decode("utf-8")
+            if digest in {data["hashes"][const.SHA256] for data in trust_data.values()}:
+                return image.digest
         return None
 
-    @staticmethod
-    def __search_image_targets_for_tag(trust_data: dict, image: Image):
-        """
-        Search in the `trust_data` for a digest, given an `image` with tag.
-        """
-        image_tag = image.tag
-        if image_tag not in trust_data:
-            return None
-
-        base64_digest = trust_data[image_tag]["hashes"]["sha256"]
-        return base64.b64decode(base64_digest).hex()
-
     async def __update_with_delegation_trust_data(
         self, trust_data, delegations, key_store, image
     ):

connaisseur/validators/notaryv1/trust_data.py

@@ -7,6 +7,7 @@
 import pytz
 from dateutil import parser
 
+import connaisseur.constants as const
 from connaisseur.exceptions import (
     InvalidTrustDataFormatError,
     NoSuchClassError,
@@ -189,7 +190,7 @@ def get_tags(self):
 
     def get_digest(self, tag: str):
         try:
-            return self.signed.get("targets", {})[tag]["hashes"]["sha256"]
+            return self.signed.get("targets", {})[tag]["hashes"][const.SHA256]
         except KeyError as err:
             msg = "Unable to find digest for tag {tag}."
             raise NotFoundException(message=msg, tag=tag) from err

tests/integration/cases.yaml

@@ -56,6 +56,13 @@ test_cases:
     namespace: default
     expected_msg: Unable to find signed digest for image docker.io/securesystemsengineering/testimage:unsigned.
     expected_result: INVALID
+  - id: rstd
+    txt: Testing signed image with tag and digest...
+    type: deploy
+    ref: securesystemsengineering/testimage:signed@sha256:c5327b291d702719a26c6cf8cc93f72e7902df46547106a9930feda2c002a4a7
+    namespace: default
+    expected_msg: pod/pod-rstd created
+    expected_result: VALID
   cosign:
   - id: cu
     txt: Testing unsigned cosign image...
@@ -78,6 +85,10 @@ test_cases:
     namespace: default
     expected_msg: pod/pod-cs created
     expected_result: null
+  - id: cstd
+    txt: Testing signed cosign image with tag and digest...
+    type: deploy
+    ref: securesystemsengineering/testimage:co-signed@sha256:c5327b291d702719a26c6cf8cc93f72e7902df46547106a9930feda2c002a4a7
   multi-cosigned:
   - id: mc-u
     txt: Testing multi-cosigned image `threshold` => undefined, not reached...

tests/test_flask_application.py

@@ -154,12 +154,12 @@ def test_create_logging_msg(msg, kwargs, out):
                     "status": {"code": 202},
                     "patchType": "JSONPatch",
                     "patch": (
-                        "W3sib3AiOiAicmVwbGFjZSIsICJwYXRoIjogI"
-                        "i9zcGVjL3RlbXBsYXRlL3NwZWMvY29udGFpbmVycy8wL2lt"
-                        "YWdlIiwgInZhbHVlIjogImRvY2tlci5pby9zZWN1cmVzeXN"
-                        "0ZW1zZW5naW5lZXJpbmcvYWxpY2UtaW1hZ2VAc2hhMjU2Om"
-                        "FjOTA0YzliMTkxZDE0ZmFmNTRiNzk1MmYyNjUwYTRiYjIxY"
-                        "zIwMWJmMzQxMzEzODhiODUxZThjZTk5MmE2NTIifV0="
+                        "W3sib3AiOiAicmVwbGFjZSIsICJwYXRoIjogIi9zcGVjL3RlbXBs"
+                        "YXRlL3NwZWMvY29udGFpbmVycy8wL2ltYWdlIiwgInZhbHVlIjog"
+                        "ImRvY2tlci5pby9zZWN1cmVzeXN0ZW1zZW5naW5lZXJpbmcvYWxp"
+                        "Y2UtaW1hZ2U6dGVzdEBzaGEyNTY6YWM5MDRjOWIxOTFkMTRmYWY1"
+                        "NGI3OTUyZjI2NTBhNGJiMjFjMjAxYmYzNDEzMTM4OGI4NTFlOGNl"
+                        "OTkyYTY1MiJ9XQ=="
                     ),
                 },
             },