fix: handle image references with both a tag and digest present

connaisseur/__main__.py

@@ -5,13 +5,12 @@
 from logging.config import dictConfig
 
 from cheroot.server import HTTPServer
-from cheroot.wsgi import Server
 from cheroot.ssl.builtin import BuiltinSSLAdapter
+from cheroot.wsgi import Server
 
 from connaisseur.flask_application import APP
 from connaisseur.logging_wrapper import ConnaisseurLoggingWrapper
 
-
 if __name__ == "__main__":
     LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO")
 

connaisseur/alert.py

@@ -7,14 +7,14 @@
 import requests
 from jinja2 import StrictUndefined, Template
 
-from connaisseur.util import safe_json_open, validate_schema
+from connaisseur.admission_request import AdmissionRequest
 from connaisseur.exceptions import (
     AlertSendingError,
     ConfigurationError,
     InvalidConfigurationFormatError,
     InvalidImageFormatError,
 )
-from connaisseur.admission_request import AdmissionRequest
+from connaisseur.util import safe_json_open, validate_schema
 
 
 class AlertingConfiguration:

connaisseur/config.py

@@ -1,6 +1,7 @@
 import collections
 import fnmatch
 import os
+
 import yaml
 
 from connaisseur.exceptions import (

connaisseur/constants.py

@@ -0,0 +1 @@
+SHA256 = "sha256"

connaisseur/flask_application.py

@@ -4,8 +4,9 @@
 import traceback
 
 from flask import Flask, jsonify, request
-from prometheus_flask_exporter import PrometheusMetrics, NO_PREFIX
+from prometheus_flask_exporter import NO_PREFIX, PrometheusMetrics
 
+import connaisseur.constants as const
 from connaisseur.admission_request import AdmissionRequest
 from connaisseur.alert import send_alerts
 from connaisseur.config import Config
@@ -16,7 +17,6 @@
 )
 from connaisseur.util import get_admission_review
 
-
 APP = Flask(__name__)
 """
 Flask application that admits the request send to the k8s cluster, validates it and
@@ -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, 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/kube_api.py

@@ -1,4 +1,5 @@
 import os
+
 import requests
 
 

connaisseur/util.py

@@ -5,7 +5,7 @@
 from typing import Optional
 
 import yaml
-from jsonschema import FormatChecker, validate, ValidationError
+from jsonschema import FormatChecker, ValidationError, validate
 
 from connaisseur.exceptions import PathTraversalError
 

connaisseur/validators/cosign/cosign_validator.py

@@ -4,20 +4,20 @@
 import os
 import re
 import subprocess  # nosec
-
 from concurrent.futures import ThreadPoolExecutor
 
+import connaisseur.constants as const
 from connaisseur.exceptions import (
     CosignError,
     CosignTimeout,
-    NotFoundException,
     InvalidFormatException,
+    NotFoundException,
     UnexpectedCosignData,
     ValidationError,
     WrongKeyError,
 )
 from connaisseur.image import Image
-from connaisseur.trust_root import KMSKey, TrustRoot, ECDSAKey
+from connaisseur.trust_root import ECDSAKey, KMSKey, TrustRoot
 from connaisseur.util import safe_path_func  # nosec
 from connaisseur.validators.interface import ValidatorInterface
 
@@ -162,7 +162,10 @@ 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 +183,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,3 +1,4 @@
+import connaisseur.constants as const
 from connaisseur.exceptions import NotFoundException
 from connaisseur.trust_root import TrustRoot
 
@@ -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,14 +7,15 @@
 import pytz
 from dateutil import parser
 
+import connaisseur.constants as const
 from connaisseur.exceptions import (
     InvalidTrustDataFormatError,
     NoSuchClassError,
     NotFoundException,
     ValidationError,
     WrongKeyError,
 )
-from connaisseur.trust_root import TrustRoot, ECDSAKey
+from connaisseur.trust_root import ECDSAKey, TrustRoot
 from connaisseur.util import validate_schema
 from connaisseur.validators.notaryv1.key_store import KeyStore
 
@@ -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

connaisseur/workload_object.py

@@ -2,7 +2,6 @@
 from connaisseur.exceptions import ParentNotFoundError, UnknownAPIVersionError
 from connaisseur.image import Image
 
-
 SUPPORTED_API_VERSIONS = {
     "Pod": ["v1"],
     "Deployment": ["apps/v1", "apps/v1beta1", "apps/v1beta2"],