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

@@ -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

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},
             }
 

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()

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:

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

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."""

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": '</v2/_catalog?n=200&last=b>; 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": '<https://reg.io/v2/_catalog?n=200>; rel="prev"'}