From 7c84822fa3a900d7d64d384f76994709dcd29c44 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pedro=20Mart=C3=ADn?= Date: Tue, 1 Sep 2026 17:18:47 +0200 Subject: [PATCH] fix(image): skip non-image OCI artifacts in registry scan (#12695) --- ...-registry-non-image-oci-artifacts.fixed.md | 1 + prowler/providers/image/image_provider.py | 101 +++++++-- prowler/providers/image/lib/registry/base.py | 9 + .../image/lib/registry/oci_adapter.py | 145 ++++++++++-- tests/providers/image/image_provider_test.py | 40 ++++ .../image/lib/registry/test_oci_adapter.py | 214 +++++++++++++++++- .../lib/registry/test_provider_registry.py | 54 +++++ 7 files changed, 522 insertions(+), 42 deletions(-) create mode 100644 prowler/changelog.d/image-registry-non-image-oci-artifacts.fixed.md diff --git a/prowler/changelog.d/image-registry-non-image-oci-artifacts.fixed.md b/prowler/changelog.d/image-registry-non-image-oci-artifacts.fixed.md new file mode 100644 index 0000000000..518c7d1926 --- /dev/null +++ b/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 diff --git a/prowler/providers/image/image_provider.py b/prowler/providers/image/image_provider.py index 8eb4f2b198..97a014d9f1 100644 --- a/prowler/providers/image/image_provider.py +++ b/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,29 +880,49 @@ class ImageProvider(Provider): # Determine if this is a Docker Hub adapter (for image reference format) is_dockerhub = isinstance(adapter, DockerHubAdapter) + skipped_artifacts = 0 + with ThreadPoolExecutor( + max_workers=min(_ENUMERATION_WORKERS, len(repositories)) + ) as pool: + all_tags = list(pool.map(adapter.list_tags, repositories)) + + 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)] + 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 in repositories: - tags = adapter.list_tags(repo) + 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("/")) + image_ref = f"{registry_host}/{repo}:{tag}" + discovered_images.append(image_ref) - # Apply tag filter - 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: - 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("/") - ) - 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: @@ -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} " diff --git a/prowler/providers/image/lib/registry/base.py b/prowler/providers/image/lib/registry/base.py index b8c08955b6..7341128cbb 100644 --- a/prowler/providers/image/lib/registry/base.py +++ b/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. diff --git a/prowler/providers/image/lib/registry/oci_adapter.py b/prowler/providers/image/lib/registry/oci_adapter.py index c5c35aef06..3d2aaf6e58 100644 --- a/prowler/providers/image/lib/registry/oci_adapter.py +++ b/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: - return - if self._basic_auth_verified: + if self._bearer_token or self._basic_auth_verified or self._anonymous_verified: return + 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: - logger.debug( - f"Bearer token rejected (HTTP 401), re-authenticating to {self.registry_url}" + 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() + 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 ) - self._bearer_token = None - self._ensure_auth() - resp = self._do_authed_request(method, url, **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)) diff --git a/tests/providers/image/image_provider_test.py b/tests/providers/image/image_provider_test.py index df80e1d35b..cb3bbf2899 100644 --- a/tests/providers/image/image_provider_test.py +++ b/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", [])] diff --git a/tests/providers/image/lib/registry/test_oci_adapter.py b/tests/providers/image/lib/registry/test_oci_adapter.py index f904c00c21..8713f98f97 100644 --- a/tests/providers/image/lib/registry/test_oci_adapter.py +++ b/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 diff --git a/tests/providers/image/lib/registry/test_provider_registry.py b/tests/providers/image/lib/registry/test_provider_registry.py index 97901ea0c4..43c8f2df4d 100644 --- a/tests/providers/image/lib/registry/test_provider_registry.py +++ b/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