feat: support tag and digest simultaneously

With this change, connaisseur now supports the use of tag 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

@@ -190,5 +190,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 = trusted_digest
         return admission_request.wl_object.get_json_patch(image, type_, index)

connaisseur/image.py

@@ -27,74 +27,63 @@ class Image:
     tag: Optional[str]
     digest: Optional[str]
 
-    def __init__(self, image: str):
-        separator = r"[-._:@+]|--"
-        alphanum = r"[A-Za-z0-9]+"
-        component = f"{alphanum}(?:(?:{separator}){alphanum})*"
-        ref = f"^{component}(?:/{component})*$"
-
-        # 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]
-
-        self.digest, self.tag = None, None
-
-        # in case the image_tag contains both a tag and a digest,
-        # the tag_re matches twice. This handles all these matches.
-        for match in re.finditer(tag_re, name_tag):
-            (digest, tag) = match.groups()
-            if digest is not None:
-                self.digest = digest
-            if tag is not None:
-                self.tag = tag
-
-        if self.tag is None and self.digest is None:
-            self.tag = "latest"
-
-        self.name = name_tag.removesuffix("@sha256:" + str(self.digest)).removesuffix(
-            ":" + str(self.tag)
-        )
-
-        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"
+    def __init__(self, image: str):  # pylint: disable=too-many-locals
+
+        # 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})*"
         )
-        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()
+        self.digest = digest.split(":")[-1] if digest else None
+        self.tag = tag or ("latest" if not self.digest else None)
+        components = name.split("/")
+        self.name = components[-1]
+        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"@sha256:{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/notaryv1/notaryv1_validator.py

@@ -50,7 +50,7 @@ async def validate(
         # search for digests or tag, depending on given image
         search_image_targets = (
             NotaryV1Validator.__search_image_targets_for_digest
-            if image.has_digest()
+            if image.digest
             else NotaryV1Validator.__search_image_targets_for_tag
         )
         # filter out the searched for digests, if present

connaisseur/validators/notaryv1/trust_data.py

@@ -136,11 +136,11 @@ def validate_hash(self, keystore: KeyStore):
             msg = "Failed to validate hash of trust data {trust_data_kind}."
             raise ValidationError(message=msg, trust_data_kind=self.kind)
 
-    def get_keys(self):
+    def get_keys(self):  # pylint: disable=no-self-use
         """Return all keys found in the trust data."""
         return {}
 
-    def get_hashes(self):
+    def get_hashes(self):  # pylint: disable=no-self-use
         """Return all hashes found in the trust data."""
         return {}
 

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,14 @@ 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=="
+                        )
                     ),
                 },
             },

tests/test_image.py

@@ -42,7 +42,7 @@
             "image",
             "tag",
             None,
-            "",
+            None,
             "registry.io",
             fix.no_exc(),
         ),
@@ -94,16 +94,7 @@
             "master-node:5000",
             fix.no_exc(),
         ),
-        ("Test/test:v1", "test", "v1", None, "", "Test", fix.no_exc()),
-        (
-            "docker.io/Library/image:tag",
-            "image",
-            "tag",
-            None,
-            None,
-            "docker.io",
-            pytest.raises(exc.InvalidImageFormatError),
-        ),
+        ("Test/test:v1", "test", "v1", None, None, "Test", fix.no_exc()),
         (
             "docker.io/library/image:Tag",
             "image",
@@ -136,48 +127,6 @@ def test_image(
         assert i.registry == registry
 
 
-@pytest.mark.parametrize(
-    "image, tag, digest",
-    [
-        (
-            "image:tag",
-            "tag",
-            "859b5aada817b3eb53410222e8fc232cf126c9e598390ae61895eb96f52ae46d",
-        )
-    ],
-)
-def test_set_digest(image: str, tag: str, digest: str):
-    i = img.Image(image)
-    i.set_digest(digest)
-    assert i.digest == digest
-    assert i.tag == tag
-
-
-@pytest.mark.parametrize(
-    "image, digest",
-    [
-        ("image:tag", False),
-        (
-            (
-                "image@sha256:859b5aada817b3eb53410222e8f"
-                "c232cf126c9e598390ae61895eb96f52ae46d"
-            ),
-            True,
-        ),
-        (
-            (
-                "image:tag@sha256:859b5aada817b3eb53410222"
-                "e8fc232cf126c9e598390ae61895eb96f52ae46d"
-            ),
-            True,
-        ),
-    ],
-)
-def test_has_digest(image: str, digest: bool):
-    i = img.Image(image)
-    assert i.has_digest() == digest
-
-
 @pytest.mark.parametrize(
     "image, str_image",
     [
@@ -205,7 +154,7 @@ def test_has_digest(image: str, digest: bool):
                 "@sha256:859b5aada817b3eb53410222e8fc232cf126c9e598390ae61895eb96f52ae46d"
             ),
             (
-                "ghcr.io/repo/test/image-with-tag-and-digest"
+                "ghcr.io/repo/test/image-with-tag-and-digest:v1.2.3"
                 "@sha256:859b5aada817b3eb53410222e8fc232cf126c9e598390ae61895eb96f52ae46d"
             ),
         ),

tests/validators/notaryv1/test_notaryv1_validator.py

@@ -85,6 +85,20 @@ def test_init(m_notary, val_config):
                 match=r"Unable to find signed digest for image.*'image': '[^']*securesystemsengineering/alice-image:missingtag",
             ),
         ),
+        (
+            "securesystemsengineering/alice-image:test@sha:ac904c9b191d14faf54b7952f2650a4bb21c201bf34131388b851e8ce992a652",
+            None,
+            [],
+            "ac904c9b191d14faf54b7952f2650a4bb21c201bf34131388b851e8ce992a652",
+            fix.no_exc(),
+        ),
+        (
+            "securesystemsengineering/alice-image:test@sha:13333333333333333333333333333337",
+            None,
+            [],
+            "",
+            pytest.raises(exc.NotFoundException, match=r".*digest.*"),
+        ),
     ],
 )
 async def test_validate(