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,61 +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]
-        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)
+    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})*"
         )
-
-        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()
+        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

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

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",
@@ -113,6 +104,15 @@
             "docker.io",
             fix.no_exc(),
         ),
+        (
+            "ghcr.io/repo/test/image-with-tag-and-digest:v1.2.3@sha256:f8816ada742348e1adfcec5c2a180b675bf6e4a294e0feb68bd70179451e1242",
+            "image-with-tag-and-digest",
+            "v1.2.3",
+            "f8816ada742348e1adfcec5c2a180b675bf6e4a294e0feb68bd70179451e1242",
+            "repo/test",
+            "ghcr.io",
+            fix.no_exc(),
+        ),
     ],
 )
 def test_image(
@@ -127,41 +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,
-        ),
-    ],
-)
-def test_has_digest(image: str, digest: bool):
-    i = img.Image(image)
-    assert i.has_digest() == digest
-
-
 @pytest.mark.parametrize(
     "image, str_image",
     [
@@ -183,6 +148,16 @@ def test_has_digest(image: str, digest: bool):
             ),
         ),
         ("path/image", "docker.io/path/image:latest"),
+        (
+            (
+                "ghcr.io/repo/test/image-with-tag-and-digest:v1.2.3"
+                "@sha256:859b5aada817b3eb53410222e8fc232cf126c9e598390ae61895eb96f52ae46d"
+            ),
+            (
+                "ghcr.io/repo/test/image-with-tag-and-digest:v1.2.3"
+                "@sha256:859b5aada817b3eb53410222e8fc232cf126c9e598390ae61895eb96f52ae46d"
+            ),
+        ),
     ],
 )
 def test_str(image: str, str_image: str):

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(