fix(image): skip non-image OCI artifacts in registry scan (#12695)

prowler/changelog.d/image-registry-non-image-oci-artifacts.fixed.md

@@ -0,0 +1 @@
+Registry scans in the Image provider now skip non-image OCI artifacts (Helm charts, cosign signatures, SBOM attestations) and no longer abort the whole scan when Trivy fails on a single discovered image; registry enumeration also runs in parallel instead of one request at a time

prowler/providers/image/image_provider.py

@@ -6,6 +6,7 @@ import re
 import subprocess
 import sys
 import tempfile
+from concurrent.futures import ThreadPoolExecutor
 from typing import Generator
 
 from alive_progress import alive_bar
@@ -49,6 +50,13 @@ from prowler.providers.image.lib.arguments.arguments import (
 from prowler.providers.image.lib.registry.dockerhub_adapter import DockerHubAdapter
 from prowler.providers.image.lib.registry.factory import create_registry_adapter
 
+# Cosign companion tags (signatures, attestations, SBOMs) — skippable by name
+# without fetching the manifest.
+COSIGN_TAG_PATTERN = re.compile(r"^sha256-[0-9a-f]{64}\.(sig|att|sbom)$")
+
+# Concurrency for registry enumeration (tag listing + manifest inspection).
+_ENUMERATION_WORKERS = 10
+
 
 class ImageProvider(Provider):
     """
@@ -164,6 +172,10 @@ class ImageProvider(Provider):
         if image_list_file:
             self._load_images_from_file(image_list_file)
 
+        # Images discovered via registry enumeration get per-image error
+        # degradation; explicitly requested images keep failing hard.
+        self._registry_discovered: set[str] = set()
+
         # Registry scan mode: enumerate images from registry
         if self.registry:
             self._enumerate_registry()
@@ -550,8 +562,15 @@ class ImageProvider(Provider):
                     for batch in self._scan_single_image(image):
                         image_findings.extend(batch)
                     yield (image, image_findings)
-                except (ImageScanError, ImageTrivyBinaryNotFoundError):
+                except ImageTrivyBinaryNotFoundError:
                     raise
+                except ImageScanError as error:
+                    if image not in self._registry_discovered:
+                        raise
+                    logger.warning(
+                        f"Skipping registry-discovered image {image}: {error}"
+                    )
+                    yield (image, [])
                 except Exception as error:
                     logger.error(f"Error scanning image {image}: {error}")
                     yield (image, [])
@@ -568,8 +587,13 @@ class ImageProvider(Provider):
         for image in self.images:
             try:
                 yield from self._scan_single_image(image)
-            except (ImageScanError, ImageTrivyBinaryNotFoundError):
+            except ImageTrivyBinaryNotFoundError:
                 raise
+            except ImageScanError as error:
+                if image not in self._registry_discovered:
+                    raise
+                logger.warning(f"Skipping registry-discovered image {image}: {error}")
+                continue
             except Exception as error:
                 logger.error(f"Error scanning image {image}: {error}")
                 continue
@@ -811,6 +835,16 @@ class ImageProvider(Provider):
             return f"Rate limited — wait or authenticate: {error_msg}"
         if any(kw in lower for kw in ("timeout", "connection refused", "no such host")):
             return f"Network issue — check connectivity: {error_msg}"
+        if any(
+            kw in lower
+            for kw in (
+                "unsupported mediatype",
+                "unsupported media type",
+                "unsupported artifact",
+                "invalid image",
+            )
+        ):
+            return f"Not a container image — trivy image cannot scan this OCI artifact: {error_msg}"
 
         return error_msg
 
@@ -846,30 +880,50 @@ class ImageProvider(Provider):
         # Determine if this is a Docker Hub adapter (for image reference format)
         is_dockerhub = isinstance(adapter, DockerHubAdapter)
 
-        discovered_images = []
-        repos_tags: dict[str, list[str]] = {}
-        for repo in repositories:
-            tags = adapter.list_tags(repo)
+        skipped_artifacts = 0
+        with ThreadPoolExecutor(
+            max_workers=min(_ENUMERATION_WORKERS, len(repositories))
+        ) as pool:
+            all_tags = list(pool.map(adapter.list_tags, repositories))
 
-            # Apply tag filter
+            candidates: list[tuple[str, str]] = []
+            for repo, tags in zip(repositories, all_tags):
                 if self._tag_filter_re:
                     tags = [t for t in tags if self._tag_filter_re.search(t)]
-
-            if tags:
-                repos_tags[repo] = tags
-
                 for tag in tags:
+                    # Cosign companions are skippable by name — no manifest fetch
+                    if COSIGN_TAG_PATTERN.match(tag):
+                        skipped_artifacts += 1
+                        continue
+                    candidates.append((repo, tag))
+
+            # Drop non-image OCI artifacts (Helm charts, signatures, SBOMs...)
+            # that trivy image cannot scan; one manifest fetch per tag, in parallel.
+            verdicts = list(
+                pool.map(lambda rt: adapter.is_container_image(*rt), candidates)
+            )
+
+        discovered_images = []
+        repos_tags: dict[str, list[str]] = {}
+        for (repo, tag), is_image in zip(candidates, verdicts):
+            if not is_image:
+                skipped_artifacts += 1
+                continue
+            repos_tags.setdefault(repo, []).append(tag)
             if is_dockerhub:
                 # Docker Hub images don't need a host prefix
                 image_ref = f"{repo}:{tag}"
             else:
                 # OCI registries need the full host/repo:tag reference
-                    registry_host = ImageProvider._strip_scheme(
-                        self.registry.rstrip("/")
-                    )
+                registry_host = ImageProvider._strip_scheme(self.registry.rstrip("/"))
                 image_ref = f"{registry_host}/{repo}:{tag}"
             discovered_images.append(image_ref)
 
+        if skipped_artifacts:
+            logger.info(
+                f"Skipped {skipped_artifacts} non-image OCI artifacts (Helm charts, signatures, SBOMs...) from registry {self.registry}"
+            )
+
         # Registry list mode: print listing and return early
         if self.registry_list_images:
             self._print_registry_listing(repos_tags, len(discovered_images))
@@ -889,6 +943,9 @@ class ImageProvider(Provider):
             if img not in existing:
                 self.images.append(img)
                 existing.add(img)
+                # Only enumeration-added images get error degradation; an image
+                # the user also requested explicitly keeps failing hard.
+                self._registry_discovered.add(img)
 
         logger.info(
             f"Discovered {len(discovered_images)} images from registry {self.registry} "

prowler/providers/image/lib/registry/base.py

@@ -132,6 +132,15 @@ class RegistryAdapter(ABC):
         """Enumerate all tags for a repository."""
         ...
 
+    def is_container_image(self, repository: str, tag: str) -> bool:
+        """Whether repository:tag points to a scannable container image.
+
+        Registries that store arbitrary OCI artifacts (Helm charts, cosign
+        signatures, SBOMs...) override this; by default everything is assumed
+        to be an image.
+        """
+        return True
+
     def _origin_url(self) -> str:
         """The URL whose host the validator compares against when enforce_origin=True.
 

prowler/providers/image/lib/registry/oci_adapter.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import base64
 import re
+import threading
 from typing import TYPE_CHECKING
 from urllib.parse import urlparse
 
@@ -19,6 +20,27 @@ if TYPE_CHECKING:
     import requests
 
 
+OCI_MANIFEST_MEDIA_TYPE = "application/vnd.oci.image.manifest.v1+json"
+
+MANIFEST_ACCEPT_TYPES = (
+    "application/vnd.docker.distribution.manifest.v2+json",
+    "application/vnd.docker.distribution.manifest.list.v2+json",
+    OCI_MANIFEST_MEDIA_TYPE,
+    "application/vnd.oci.image.index.v1+json",
+)
+
+# Multi-arch indexes: their children are resolved by trivy itself.
+INDEX_MEDIA_TYPES = (
+    "application/vnd.docker.distribution.manifest.list.v2+json",
+    "application/vnd.oci.image.index.v1+json",
+)
+
+IMAGE_CONFIG_MEDIA_TYPES = (
+    "application/vnd.oci.image.config.v1+json",
+    "application/vnd.docker.container.image.v1+json",
+)
+
+
 class OciRegistryAdapter(RegistryAdapter):
     """Adapter for registries implementing OCI Distribution Spec."""
 
@@ -34,6 +56,19 @@ class OciRegistryAdapter(RegistryAdapter):
         self._base_url = self._normalise_url(registry_url)
         self._bearer_token: str | None = None
         self._basic_auth_verified = False
+        self._anonymous_verified = False
+        # Enumeration inspects manifests from a thread pool; serialise token
+        # exchanges so N concurrent 401s don't trigger N auth round-trips.
+        self._auth_lock = threading.Lock()
+
+    def __getstate__(self) -> dict:
+        state = super().__getstate__()
+        del state["_auth_lock"]
+        return state
+
+    def __setstate__(self, state: dict) -> None:
+        self.__dict__.update(state)
+        self._auth_lock = threading.Lock()
 
     @staticmethod
     def _normalise_url(url: str) -> str:
@@ -94,17 +129,78 @@ class OciRegistryAdapter(RegistryAdapter):
             params = {}
         return tags
 
+    def is_container_image(self, repository: str, tag: str) -> bool:
+        """Inspect the manifest to tell container images apart from other OCI artifacts.
+
+        Uncertainty (network error, unparseable manifest) resolves to True so a
+        transient failure never silently drops a real image — trivy gives the
+        final verdict.
+        """
+        url = f"{self._base_url}/v2/{repository}/manifests/{tag}"
+        try:
+            self._ensure_auth(repository=repository)
+            resp = self._authed_request(
+                "GET", url, headers={"Accept": ", ".join(MANIFEST_ACCEPT_TYPES)}
+            )
+        except Exception as error:
+            logger.warning(
+                f"Could not fetch manifest for {repository}:{tag}, assuming image: {error}"
+            )
+            return True
+        if resp.status_code != 200:
+            logger.warning(
+                f"Manifest request for {repository}:{tag} returned HTTP {resp.status_code}, assuming image"
+            )
+            return True
+
+        # RFC 9110: media type tokens are case-insensitive
+        content_type = (
+            resp.headers.get("Content-Type", "").split(";")[0].strip().lower()
+        )
+        if content_type == "application/vnd.docker.distribution.manifest.v2+json":
+            return True
+        if content_type in INDEX_MEDIA_TYPES or content_type == OCI_MANIFEST_MEDIA_TYPE:
+            # Helm charts, cosign signatures, SBOMs... reuse the OCI manifest
+            # media type, and since image-spec v1.1 an index can also represent
+            # a non-image artifact; only artifactType (or, for manifests,
+            # config.mediaType) tells them apart.
+            try:
+                manifest = resp.json()
+            except ValueError:
+                return True
+            artifact_type = (manifest.get("artifactType") or "").lower()
+            if artifact_type:
+                return artifact_type in IMAGE_CONFIG_MEDIA_TYPES
+            if content_type in INDEX_MEDIA_TYPES:
+                # Plain multi-arch index
+                return True
+            config_type = (manifest.get("config", {}).get("mediaType") or "").lower()
+            return config_type in IMAGE_CONFIG_MEDIA_TYPES
+        logger.info(
+            f"Skipping {repository}:{tag} — manifest media type {content_type or 'unknown'} is not a container image"
+        )
+        return False
+
     def _ensure_auth(self, repository: str | None = None) -> None:
-        if self._bearer_token:
+        if self._bearer_token or self._basic_auth_verified or self._anonymous_verified:
             return
-        if self._basic_auth_verified:
+        with self._auth_lock:
+            if (
+                self._bearer_token
+                or self._basic_auth_verified
+                or self._anonymous_verified
+            ):
                 return
+            self._authenticate(repository=repository)
+
+    def _authenticate(self, repository: str | None = None) -> None:
         if self.token:
             self._bearer_token = self.token
             return
         ping_url = f"{self._base_url}/v2/"
         resp = self._request_with_retry("GET", ping_url)
         if resp.status_code == 200:
+            self._anonymous_verified = True
             return
         if resp.status_code == 401:
             www_auth = resp.headers.get("Www-Authenticate", "")
@@ -206,12 +302,29 @@ class OciRegistryAdapter(RegistryAdapter):
     def _authed_request(self, method: str, url: str, **kwargs) -> requests.Response:
         resp = self._do_authed_request(method, url, **kwargs)
         if resp.status_code == 401 and self._bearer_token:
+            challenge = self._find_challenge(
+                resp.headers.get("Www-Authenticate", ""), "Bearer"
+            )
+            if challenge and self._is_same_origin_as_registry(url):
+                # The cached token may be scoped to another repository; the
+                # response challenge names the exact scope this endpoint needs.
+                logger.debug(
+                    f"Bearer token rejected (HTTP 401), re-authenticating with response challenge scope for {url}"
+                )
+                fresh_token = self._obtain_bearer_token(challenge)
+                self._bearer_token = fresh_token
+            else:
                 logger.debug(
                     f"Bearer token rejected (HTTP 401), re-authenticating to {self.registry_url}"
                 )
                 self._bearer_token = None
                 self._ensure_auth()
-            resp = self._do_authed_request(method, url, **kwargs)
+                fresh_token = self._bearer_token
+            # Retry with the token this request obtained: a concurrent worker
+            # may have already replaced the shared one with another scope.
+            resp = self._do_authed_request(
+                method, url, bearer_token=fresh_token, **kwargs
+            )
         if (
             resp.status_code == 401
             and self._bearer_token
@@ -241,15 +354,21 @@ class OciRegistryAdapter(RegistryAdapter):
                 # this endpoint demands Bearer. The challenge carries the right
                 # scope.
                 logger.debug(f"Basic auth not accepted for {url}, switching to Bearer")
-                self._bearer_token = self._obtain_bearer_token(bearer_challenge)
-                resp = self._do_authed_request(method, url, **kwargs)
+                fresh_token = self._obtain_bearer_token(bearer_challenge)
+                self._bearer_token = fresh_token
+                resp = self._do_authed_request(
+                    method, url, bearer_token=fresh_token, **kwargs
+                )
         return resp
 
-    def _do_authed_request(self, method: str, url: str, **kwargs) -> requests.Response:
+    def _do_authed_request(
+        self, method: str, url: str, bearer_token: str | None = None, **kwargs
+    ) -> requests.Response:
         headers = kwargs.pop("headers", {})
         if self._is_same_origin_as_registry(url):
-            if self._bearer_token:
-                headers["Authorization"] = f"Bearer {self._bearer_token}"
+            token = bearer_token or self._bearer_token
+            if token:
+                headers["Authorization"] = f"Bearer {token}"
             elif self.username and self.password:
                 user, pwd = self._resolve_basic_credentials()
                 kwargs.setdefault("auth", (user, pwd))

tests/providers/image/image_provider_test.py

@@ -1452,3 +1452,43 @@ class TestConnectionPrivateNetworkAllowlist:
             result = ImageProvider.test_connection(image="harbor.internal")
 
         assert result.is_connected is True
+
+
+class TestRegistryScanErrorDegradation:
+    @patch("subprocess.run")
+    def test_registry_discovered_image_scan_error_is_skipped(self, mock_subprocess):
+        provider = _make_provider(images=["reg.io/chart:1.0", "alpine:3.18"])
+        provider._registry_discovered = {"reg.io/chart:1.0"}
+        mock_subprocess.side_effect = [
+            MagicMock(returncode=1, stdout="", stderr="unsupported media type"),
+            MagicMock(returncode=0, stdout=get_sample_trivy_json_output(), stderr=""),
+        ]
+
+        reports = []
+        for batch in provider.run_scan():
+            reports.extend(batch)
+
+        assert len(reports) == 1
+        assert reports[0].check_metadata.CheckID == "CVE-2024-1234"
+
+    @patch("subprocess.run")
+    def test_explicit_image_scan_error_still_raises(self, mock_subprocess):
+        provider = _make_provider(images=["alpine:3.18"])
+        mock_subprocess.return_value = MagicMock(
+            returncode=1, stdout="", stderr="unsupported media type"
+        )
+
+        with pytest.raises(ImageScanError):
+            for _ in provider.run_scan():
+                pass
+
+    @patch("subprocess.run")
+    def test_scan_per_image_degrades_registry_discovered_error(self, mock_subprocess):
+        provider = _make_provider(images=["reg.io/chart:1.0"])
+        provider._registry_discovered = {"reg.io/chart:1.0"}
+        mock_subprocess.return_value = MagicMock(
+            returncode=1, stdout="", stderr="unsupported media type"
+        )
+
+        results = list(provider.scan_per_image())
+        assert results == [("reg.io/chart:1.0", [])]

tests/providers/image/lib/registry/test_oci_adapter.py

@@ -225,8 +225,8 @@ class TestOciAdapterAuth:
     def test_authed_request_retries_on_401_with_bearer(self, mock_request):
         adapter = OciRegistryAdapter("reg.io", username="u", password="p")
         adapter._bearer_token = "expired-token"
-        # First request: 401 (expired token)
-        resp_401 = MagicMock(status_code=401)
+        # First request: 401 without a challenge -> blind re-auth fallback
+        resp_401 = MagicMock(status_code=401, headers={})
         # _ensure_auth ping: 401 with bearer challenge
         ping_resp = MagicMock(
             status_code=401,
@@ -905,11 +905,10 @@ class TestBasicAuthFallback:
         mock_request.side_effect = [
             ping,
             token,
-            catalog_401,
-            ping,
-            token,
-            catalog_401,
-            catalog_ok,
+            catalog_401,  # bearer without catalog scope
+            token,  # re-auth straight from the response's Bearer challenge
+            catalog_401,  # scoped bearer retry, same result
+            catalog_ok,  # basic fallback
         ]
 
         adapter = OciRegistryAdapter("reg.io", username="admin", password="secret")
@@ -1125,3 +1124,204 @@ class TestBearerAuthSwitch:
         adapter = OciRegistryAdapter("reg.io", username="admin", password="wrong")
         with pytest.raises(ImageRegistryAuthError, match="bearer token"):
             adapter.list_repositories()
+
+
+class TestOciAdapterIsContainerImage:
+    def _adapter(self):
+        return OciRegistryAdapter("reg.io", token="t")
+
+    @staticmethod
+    def _manifest_resp(content_type, body=None):
+        resp = MagicMock(status_code=200, headers={"Content-Type": content_type})
+        resp.json.return_value = body or {}
+        return resp
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_docker_v2_manifest_is_image(self, mock_request):
+        mock_request.return_value = self._manifest_resp(
+            "application/vnd.docker.distribution.manifest.v2+json"
+        )
+        assert self._adapter().is_container_image("app", "1.0") is True
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_oci_index_is_image(self, mock_request):
+        mock_request.return_value = self._manifest_resp(
+            "application/vnd.oci.image.index.v1+json"
+        )
+        assert self._adapter().is_container_image("app", "1.0") is True
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_oci_manifest_with_image_config_is_image(self, mock_request):
+        mock_request.return_value = self._manifest_resp(
+            "application/vnd.oci.image.manifest.v1+json",
+            {"config": {"mediaType": "application/vnd.oci.image.config.v1+json"}},
+        )
+        assert self._adapter().is_container_image("app", "1.0") is True
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_helm_chart_is_not_image(self, mock_request):
+        mock_request.return_value = self._manifest_resp(
+            "application/vnd.oci.image.manifest.v1+json",
+            {"config": {"mediaType": "application/vnd.cncf.helm.config.v1+json"}},
+        )
+        assert self._adapter().is_container_image("charts/app", "1.0") is False
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_artifact_type_wins_over_config(self, mock_request):
+        mock_request.return_value = self._manifest_resp(
+            "application/vnd.oci.image.manifest.v1+json",
+            {
+                "artifactType": "application/vnd.dev.cosign.artifact.sig.v1+json",
+                "config": {"mediaType": "application/vnd.oci.image.config.v1+json"},
+            },
+        )
+        assert self._adapter().is_container_image("app", "sig") is False
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_unknown_media_type_is_not_image(self, mock_request):
+        mock_request.return_value = self._manifest_resp(
+            "application/vnd.cncf.helm.chart.content.v1.tar+gzip"
+        )
+        assert self._adapter().is_container_image("charts/app", "1.0") is False
+
+    @patch("prowler.providers.image.lib.registry.base.time.sleep")
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_manifest_error_assumes_image(self, mock_request, _mock_sleep):
+        mock_request.side_effect = requests.exceptions.ConnectionError("boom")
+        assert self._adapter().is_container_image("app", "1.0") is True
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_manifest_404_assumes_image(self, mock_request):
+        mock_request.return_value = MagicMock(status_code=404, headers={})
+        assert self._adapter().is_container_image("app", "1.0") is True
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_anonymous_auth_pings_only_once(self, mock_request):
+        mock_request.return_value = MagicMock(status_code=200, headers={})
+        adapter = OciRegistryAdapter("reg.io")
+        adapter._ensure_auth()
+        adapter._ensure_auth(repository="app")
+        adapter._ensure_auth(repository="other")
+        assert mock_request.call_count == 1
+
+
+class TestOciAdapterScopedReauth:
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_401_with_challenge_reauths_with_response_scope(self, mock_request):
+        adapter = OciRegistryAdapter("reg.io", username="u", password="p")
+        adapter._bearer_token = "token-scoped-to-other-repo"
+        resp_401 = MagicMock(
+            status_code=401,
+            headers={
+                "Www-Authenticate": 'Bearer realm="https://auth.reg.io/token",service="registry",scope="repository:myapp:pull"'
+            },
+        )
+        token_resp = MagicMock(status_code=200)
+        token_resp.json.return_value = {"token": "myapp-scoped-token"}
+        resp_200 = MagicMock(status_code=200)
+        mock_request.side_effect = [resp_401, token_resp, resp_200]
+
+        result = adapter._authed_request(
+            "GET", "https://reg.io/v2/myapp/manifests/latest"
+        )
+
+        assert result.status_code == 200
+        assert adapter._bearer_token == "myapp-scoped-token"
+        # Token exchange used the scope from the response challenge, no blind /v2/ ping
+        token_call = mock_request.call_args_list[1]
+        assert token_call.kwargs["params"]["scope"] == "repository:myapp:pull"
+        assert mock_request.call_count == 3
+
+
+class TestOciAdapterPickle:
+    def test_adapter_is_picklable_despite_auth_lock(self):
+        import pickle
+
+        adapter = OciRegistryAdapter("reg.io", username="u", password="p")
+        restored = pickle.loads(pickle.dumps(adapter))
+        assert restored._base_url == "https://reg.io"
+        # The lock is recreated, not carried over
+        assert restored._auth_lock is not adapter._auth_lock
+        restored._ensure_auth  # attribute access must not blow up
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_retry_uses_own_token_despite_concurrent_overwrite(self, mock_request):
+        adapter = OciRegistryAdapter("reg.io", username="u", password="p")
+        adapter._bearer_token = "stale-token"
+        resp_401 = MagicMock(
+            status_code=401,
+            headers={
+                "Www-Authenticate": 'Bearer realm="https://auth.reg.io/token",service="registry",scope="repository:myapp:pull"'
+            },
+        )
+        token_resp = MagicMock(status_code=200)
+        token_resp.json.return_value = {"token": "fresh-token"}
+        resp_200 = MagicMock(status_code=200)
+        mock_request.side_effect = [resp_401, token_resp, resp_200]
+
+        # Another worker replaces the shared token right after this request's
+        # token exchange
+        original = adapter._obtain_bearer_token
+
+        def clobbering(challenge, repository=None):
+            token = original(challenge, repository)
+            adapter._bearer_token = "other-repo-token"
+            return token
+
+        adapter._obtain_bearer_token = clobbering
+
+        result = adapter._authed_request(
+            "GET", "https://reg.io/v2/myapp/manifests/latest"
+        )
+
+        assert result.status_code == 200
+        retry_headers = mock_request.call_args_list[2].kwargs["headers"]
+        assert retry_headers["Authorization"] == "Bearer fresh-token"
+
+
+class TestOciAdapterArtifactIndexAndCaseInsensitivity:
+    def _adapter(self):
+        return OciRegistryAdapter("reg.io", token="t")
+
+    @staticmethod
+    def _manifest_resp(content_type, body=None):
+        resp = MagicMock(status_code=200, headers={"Content-Type": content_type})
+        resp.json.return_value = body or {}
+        return resp
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_artifact_index_is_not_image(self, mock_request):
+        # image-spec v1.1: an index can represent a non-image artifact
+        mock_request.return_value = self._manifest_resp(
+            "application/vnd.oci.image.index.v1+json",
+            {
+                "artifactType": "application/vnd.cncf.helm.config.v1+json",
+                "manifests": [],
+            },
+        )
+        assert self._adapter().is_container_image("charts/app", "1.0") is False
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_multiarch_index_without_artifact_type_is_image(self, mock_request):
+        mock_request.return_value = self._manifest_resp(
+            "application/vnd.oci.image.index.v1+json",
+            {"schemaVersion": 2, "manifests": [{"platform": {"os": "linux"}}]},
+        )
+        assert self._adapter().is_container_image("app", "latest") is True
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_mixed_case_media_type_is_normalized(self, mock_request):
+        # RFC 9110: type/subtype are case-insensitive
+        mock_request.return_value = self._manifest_resp(
+            "Application/vnd.OCI.Image.Manifest.v1+JSON",
+            {"config": {"mediaType": "application/vnd.oci.image.config.v1+json"}},
+        )
+        assert self._adapter().is_container_image("app", "1.0") is True
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_mixed_case_artifact_type_still_rejected(self, mock_request):
+        mock_request.return_value = self._manifest_resp(
+            "application/vnd.oci.image.manifest.v1+json",
+            {"artifactType": "Application/vnd.CNCF.Helm.Config.v1+json"},
+        )
+        assert self._adapter().is_container_image("charts/app", "1.0") is False

tests/providers/image/lib/registry/test_provider_registry.py

@@ -232,3 +232,57 @@ class TestDockerHubEnumeration:
         for img in provider.images:
             assert not img.startswith("docker.io/"), f"Unexpected host prefix in {img}"
         assert len(provider.images) == 3
+
+
+class TestNonImageArtifactFiltering:
+    @patch("prowler.providers.image.image_provider.create_registry_adapter")
+    def test_cosign_tags_skipped_without_manifest_fetch(self, mock_factory):
+        adapter = MagicMock()
+        adapter.list_repositories.return_value = ["app"]
+        adapter.list_tags.return_value = [
+            "latest",
+            "sha256-" + "a" * 64 + ".sig",
+            "sha256-" + "b" * 64 + ".att",
+            "sha256-" + "c" * 64 + ".sbom",
+        ]
+        adapter.is_container_image.return_value = True
+        mock_factory.return_value = adapter
+
+        provider = _build_provider()
+        assert provider.images == ["myregistry.io/app:latest"]
+        adapter.is_container_image.assert_called_once_with("app", "latest")
+
+    @patch("prowler.providers.image.image_provider.create_registry_adapter")
+    def test_non_image_artifacts_skipped(self, mock_factory):
+        adapter = MagicMock()
+        adapter.list_repositories.return_value = ["app", "charts/app"]
+        adapter.list_tags.return_value = ["1.0"]
+        adapter.is_container_image.side_effect = lambda repo, _tag: repo == "app"
+        mock_factory.return_value = adapter
+
+        provider = _build_provider()
+        assert provider.images == ["myregistry.io/app:1.0"]
+
+    @patch("prowler.providers.image.image_provider.create_registry_adapter")
+    def test_discovered_images_tracked_for_error_degradation(self, mock_factory):
+        adapter = MagicMock()
+        adapter.list_repositories.return_value = ["app"]
+        adapter.list_tags.return_value = ["latest"]
+        adapter.is_container_image.return_value = True
+        mock_factory.return_value = adapter
+
+        provider = _build_provider(images=["nginx:latest"])
+        assert "myregistry.io/app:latest" in provider._registry_discovered
+        assert "nginx:latest" not in provider._registry_discovered
+
+    @patch("prowler.providers.image.image_provider.create_registry_adapter")
+    def test_explicit_image_also_discovered_keeps_hard_failure(self, mock_factory):
+        adapter = MagicMock()
+        adapter.list_repositories.return_value = ["myapp"]
+        adapter.list_tags.return_value = ["latest"]
+        adapter.is_container_image.return_value = True
+        mock_factory.return_value = adapter
+
+        provider = _build_provider(images=["myregistry.io/myapp:latest"])
+        # The user asked for it explicitly: no error degradation
+        assert "myregistry.io/myapp:latest" not in provider._registry_discovered