From 18bac5a89012a5d9e0bfcb7a58beeea345bd2e71 Mon Sep 17 00:00:00 2001 From: "Andoni A." <14891798+andoniaf@users.noreply.github.com> Date: Mon, 23 Feb 2026 08:04:51 +0100 Subject: [PATCH] fix: address PR review comments for Image provider - Extract _is_registry_url() into ImageProvider to deduplicate logic - Restore relative URL handling in _next_page_url for path-only Link headers - Yield empty findings on generic error in scan_per_image - Add credential validation to ImageProviderSecret serializer --- api/src/backend/api/tests/test_serializers.py | 35 +++++++++++++++++++ api/src/backend/api/utils.py | 18 ++-------- api/src/backend/api/v1/serializers.py | 15 ++++++++ prowler/providers/image/image_provider.py | 35 +++++++++---------- prowler/providers/image/lib/registry/base.py | 7 +++- tests/providers/image/image_provider_test.py | 29 +++++++++++++++ .../image/lib/registry/test_oci_adapter.py | 10 ++++++ 7 files changed, 115 insertions(+), 34 deletions(-) diff --git a/api/src/backend/api/tests/test_serializers.py b/api/src/backend/api/tests/test_serializers.py index a52b3464d8..5810a97b63 100644 --- a/api/src/backend/api/tests/test_serializers.py +++ b/api/src/backend/api/tests/test_serializers.py @@ -2,6 +2,7 @@ import pytest from rest_framework.exceptions import ValidationError from api.v1.serializer_utils.integrations import S3ConfigSerializer +from api.v1.serializers import ImageProviderSecret class TestS3ConfigSerializer: @@ -98,3 +99,37 @@ class TestS3ConfigSerializer: serializer = S3ConfigSerializer(data=data) assert not serializer.is_valid() assert "output_directory" in serializer.errors + + +class TestImageProviderSecret: + """Test cases for ImageProviderSecret validation.""" + + def test_valid_no_credentials(self): + serializer = ImageProviderSecret(data={}) + assert serializer.is_valid() + + def test_valid_token_only(self): + serializer = ImageProviderSecret(data={"registry_token": "tok"}) + assert serializer.is_valid() + + def test_valid_username_and_password(self): + serializer = ImageProviderSecret( + data={"registry_username": "user", "registry_password": "pass"} + ) + assert serializer.is_valid() + + def test_valid_token_with_username_only(self): + serializer = ImageProviderSecret( + data={"registry_token": "tok", "registry_username": "user"} + ) + assert serializer.is_valid() + + def test_invalid_username_without_password(self): + serializer = ImageProviderSecret(data={"registry_username": "user"}) + assert not serializer.is_valid() + assert "non_field_errors" in serializer.errors + + def test_invalid_password_without_username(self): + serializer = ImageProviderSecret(data={"registry_password": "pass"}) + assert not serializer.is_valid() + assert "non_field_errors" in serializer.errors diff --git a/api/src/backend/api/utils.py b/api/src/backend/api/utils.py index 848466faff..4a6e4759b6 100644 --- a/api/src/backend/api/utils.py +++ b/api/src/backend/api/utils.py @@ -230,28 +230,16 @@ def get_prowler_provider_kwargs( elif provider.provider == Provider.ProviderChoices.IMAGE.value: # Detect whether uid is a registry URL (e.g. "docker.io/andoniaf") or # a concrete image reference (e.g. "docker.io/andoniaf/myimage:latest"). - # Uses the same heuristic as ImageProvider.test_connection. from prowler.providers.image.image_provider import ImageProvider - image_uid = provider.uid - registry_host = ImageProvider._extract_registry(image_uid) - if registry_host: - repo_and_tag = image_uid[len(registry_host) + 1 :] - else: - repo_and_tag = image_uid - - is_registry_url = ( - registry_host and "/" not in repo_and_tag and ":" not in repo_and_tag - ) - - if is_registry_url: + if ImageProvider._is_registry_url(provider.uid): prowler_provider_kwargs = { - "registry": image_uid, + "registry": provider.uid, **{k: v for k, v in prowler_provider_kwargs.items() if v}, } else: prowler_provider_kwargs = { - "images": [image_uid], + "images": [provider.uid], **{k: v for k, v in prowler_provider_kwargs.items() if v}, } diff --git a/api/src/backend/api/v1/serializers.py b/api/src/backend/api/v1/serializers.py index 0a40aeba82..55ed92174a 100644 --- a/api/src/backend/api/v1/serializers.py +++ b/api/src/backend/api/v1/serializers.py @@ -1712,6 +1712,21 @@ class ImageProviderSecret(serializers.Serializer): class Meta: resource_name = "provider-secrets" + def validate(self, attrs): + token = attrs.get("registry_token") + username = attrs.get("registry_username") + password = attrs.get("registry_password") + if not token: + if username and not password: + raise serializers.ValidationError( + "registry_password is required when registry_username is provided." + ) + if password and not username: + raise serializers.ValidationError( + "registry_username is required when registry_password is provided." + ) + return attrs + class AlibabaCloudProviderSecret(serializers.Serializer): access_key_id = serializers.CharField() diff --git a/prowler/providers/image/image_provider.py b/prowler/providers/image/image_provider.py index c0ad68d855..9817656dd2 100644 --- a/prowler/providers/image/image_provider.py +++ b/prowler/providers/image/image_provider.py @@ -325,6 +325,19 @@ class ImageProvider(Provider): return parts[0] return None + @staticmethod + def _is_registry_url(image_uid: str) -> bool: + """Determine whether an image UID is a registry URL (namespace only). + + A registry URL like ``docker.io/andoniaf`` has a registry host but + the remaining part contains no ``/`` (no repo) and no ``:`` (no tag). + """ + registry_host = ImageProvider._extract_registry(image_uid) + if not registry_host: + return False + repo_and_tag = image_uid[len(registry_host) + 1 :] + return "/" not in repo_and_tag and ":" not in repo_and_tag + def cleanup(self) -> None: """Clean up any resources after scanning.""" @@ -488,7 +501,7 @@ class ImageProvider(Provider): raise except Exception as error: logger.error(f"Error scanning image {image}: {error}") - continue + yield (image, []) finally: self.cleanup() @@ -933,23 +946,7 @@ class ImageProvider(Provider): if not image: return Connection(is_connected=False, error="Image name is required") - # Parse registry, repository, and tag from image reference - registry_host = ImageProvider._extract_registry(image) - - if registry_host: - repo_and_tag = image[len(registry_host) + 1 :] - else: - repo_and_tag = image - - # Determine if this is a registry URL (namespace only) or a full - # image reference. A registry URL like ``docker.io/andoniaf`` has - # a registry host but the remaining part contains no ``/`` (no - # repo) and no ``:`` (no tag). - is_registry_url = ( - registry_host and "/" not in repo_and_tag and ":" not in repo_and_tag - ) - - if is_registry_url: + if ImageProvider._is_registry_url(image): # Registry enumeration mode — test by listing repositories adapter = create_registry_adapter( registry_url=image, @@ -961,6 +958,8 @@ class ImageProvider(Provider): return Connection(is_connected=True) # Image reference mode — verify the specific tag exists + registry_host = ImageProvider._extract_registry(image) + repo_and_tag = image[len(registry_host) + 1 :] if registry_host else image if ":" in repo_and_tag: repository, tag = repo_and_tag.rsplit(":", 1) else: diff --git a/prowler/providers/image/lib/registry/base.py b/prowler/providers/image/lib/registry/base.py index 133ef9d4e0..1bb26ccbfb 100644 --- a/prowler/providers/image/lib/registry/base.py +++ b/prowler/providers/image/lib/registry/base.py @@ -5,6 +5,7 @@ from __future__ import annotations import re import time from abc import ABC, abstractmethod +from urllib.parse import urlparse import requests @@ -137,5 +138,9 @@ class RegistryAdapter(ABC): return None match = re.search(r'<([^>]+)>;\s*rel="next"', link_header) if match: - return match.group(1) + url = match.group(1) + if url.startswith("/"): + parsed = urlparse(resp.url) + return f"{parsed.scheme}://{parsed.netloc}{url}" + return url return None diff --git a/tests/providers/image/image_provider_test.py b/tests/providers/image/image_provider_test.py index 8d1dc0e6d3..b25e9edfc2 100644 --- a/tests/providers/image/image_provider_test.py +++ b/tests/providers/image/image_provider_test.py @@ -697,6 +697,35 @@ class TestExtractRegistry: assert ImageProvider._extract_registry("nginx") is None +class TestIsRegistryUrl: + def test_registry_url_with_namespace(self): + assert ImageProvider._is_registry_url("docker.io/andoniaf") is True + + def test_registry_url_ghcr(self): + assert ImageProvider._is_registry_url("ghcr.io/org") is True + + def test_image_ref_with_tag(self): + assert ImageProvider._is_registry_url("ghcr.io/user/image:tag") is False + + def test_image_ref_with_repo(self): + assert ImageProvider._is_registry_url("ghcr.io/user/image") is False + + def test_dockerhub_short_image(self): + assert ImageProvider._is_registry_url("alpine:3.18") is False + + def test_dockerhub_with_namespace(self): + assert ImageProvider._is_registry_url("andoniaf/test:tag") is False + + def test_bare_image_name(self): + assert ImageProvider._is_registry_url("nginx") is False + + def test_localhost_namespace(self): + assert ImageProvider._is_registry_url("localhost:5000/myns") is True + + def test_localhost_image_with_tag(self): + assert ImageProvider._is_registry_url("localhost:5000/myns/image:v1") is False + + class TestCleanup: def test_cleanup_idempotent(self): """Test cleanup is safe to call multiple times.""" diff --git a/tests/providers/image/lib/registry/test_oci_adapter.py b/tests/providers/image/lib/registry/test_oci_adapter.py index 32b0202e2d..b1006ea6cd 100644 --- a/tests/providers/image/lib/registry/test_oci_adapter.py +++ b/tests/providers/image/lib/registry/test_oci_adapter.py @@ -300,6 +300,16 @@ class TestOciAdapterNextPageUrl: == "https://reg.io/v2/_catalog?n=200&last=b" ) + def test_link_header_relative_url(self): + resp = MagicMock( + headers={"Link": '; rel="next"'}, + url="https://reg.io/v2/_catalog?n=200", + ) + assert ( + OciRegistryAdapter._next_page_url(resp) + == "https://reg.io/v2/_catalog?n=200&last=b" + ) + def test_link_header_no_next(self): resp = MagicMock( headers={"Link": '; rel="prev"'}