fix(sdk): use registry HTTP check for Image provider connection test

- Replace Trivy subprocess with registry adapter for connection testing
- Avoid false auth failures caused by Trivy DB download errors
- Update tests to mock registry adapters instead of subprocess

prowler/providers/image/image_provider.py

@@ -31,6 +31,9 @@ from prowler.providers.image.exceptions.exceptions import (
     ImageListFileReadError,
     ImageMaxImagesExceededError,
     ImageNoImagesProvidedError,
+    ImageRegistryAuthError,
+    ImageRegistryCatalogError,
+    ImageRegistryNetworkError,
     ImageScanError,
     ImageTrivyBinaryNotFoundError,
 )
@@ -901,7 +904,10 @@ class ImageProvider(Provider):
         registry_token: str | None = None,
     ) -> "Connection":
         """
-        Test connection to container registry by attempting to inspect an image.
+        Test connection to container registry by verifying image accessibility.
+
+        Uses registry HTTP APIs directly instead of Trivy to avoid false
+        failures caused by Trivy DB download issues.
 
         Args:
             image: Container image to test
@@ -921,61 +927,59 @@ class ImageProvider(Provider):
             if not image:
                 return Connection(is_connected=False, error="Image name is required")
 
-            # Build env with registry credentials
-            env = dict(os.environ)
-            if registry_username and registry_password:
-                env["TRIVY_USERNAME"] = registry_username
-                env["TRIVY_PASSWORD"] = registry_password
-            elif registry_token:
-                env["TRIVY_REGISTRY_TOKEN"] = registry_token
+            # Parse registry, repository, and tag from image reference
+            registry_host = ImageProvider._extract_registry(image)
 
-            # Test by scanning the image with a short timeout; Trivy will
-            # download the vulnerability DB on the first run if needed.
-            process = subprocess.run(
-                [
-                    "trivy",
-                    "image",
-                    "--format",
-                    "json",
-                    "--timeout",
-                    "5m",
-                    image,
-                ],
-                capture_output=True,
-                text=True,
-                timeout=360,
-                env=env,
-            )
-
-            if process.returncode == 0:
-                return Connection(is_connected=True)
+            if registry_host:
+                repo_and_tag = image[len(registry_host) + 1 :]
             else:
-                error_msg = process.stderr or "Unknown error"
-                if "401" in error_msg or "unauthorized" in error_msg.lower():
-                    return Connection(
-                        is_connected=False,
-                        error="Authentication failed. Check registry credentials.",
-                    )
-                elif "not found" in error_msg.lower() or "404" in error_msg:
-                    return Connection(
-                        is_connected=False,
-                        error="Image not found in registry.",
-                    )
-                else:
-                    return Connection(
-                        is_connected=False,
-                        error=f"Failed to access image: {error_msg[:200]}",
-                    )
+                repo_and_tag = image
 
-        except subprocess.TimeoutExpired:
-            return Connection(
-                is_connected=False,
-                error="Connection timed out",
+            if ":" in repo_and_tag:
+                repository, tag = repo_and_tag.rsplit(":", 1)
+            else:
+                repository = repo_and_tag
+                tag = "latest"
+
+            is_dockerhub = not registry_host or registry_host in (
+                "docker.io",
+                "registry-1.docker.io",
             )
-        except FileNotFoundError:
+
+            # Docker Hub official images use "library/" prefix
+            if is_dockerhub and "/" not in repository:
+                repository = f"library/{repository}"
+
+            if is_dockerhub:
+                registry_url = f"docker.io/{repository.split('/')[0]}"
+            else:
+                registry_url = registry_host
+
+            adapter = create_registry_adapter(
+                registry_url=registry_url,
+                username=registry_username,
+                password=registry_password,
+                token=registry_token,
+            )
+
+            tags = adapter.list_tags(repository)
+            if tag not in tags:
+                return Connection(
+                    is_connected=False,
+                    error=f"Tag '{tag}' not found for image '{image}'.",
+                )
+
+            return Connection(is_connected=True)
+
+        except ImageRegistryAuthError:
             return Connection(
                 is_connected=False,
-                error="Trivy binary not found. Please install Trivy.",
+                error="Authentication failed. Check registry credentials.",
+            )
+        except (ImageRegistryNetworkError, ImageRegistryCatalogError) as exc:
+            return Connection(
+                is_connected=False,
+                error=f"Failed to access image: {str(exc)[:200]}",
             )
         except Exception as error:
             if raise_on_exception:

tests/providers/image/image_provider_test.py

@@ -15,6 +15,7 @@ from prowler.providers.image.exceptions.exceptions import (
     ImageListFileNotFoundError,
     ImageListFileReadError,
     ImageNoImagesProvidedError,
+    ImageRegistryAuthError,
     ImageScanError,
     ImageTrivyBinaryNotFoundError,
 )
@@ -288,20 +289,23 @@ class TestImageProvider:
             )
             assert "alpine:3.18" in output
 
-    @patch("subprocess.run")
-    def test_test_connection_success(self, mock_subprocess):
+    @patch("prowler.providers.image.image_provider.create_registry_adapter")
+    def test_test_connection_success(self, mock_factory):
         """Test successful connection returns is_connected=True."""
-        mock_subprocess.return_value = MagicMock(returncode=0, stderr="")
+        mock_adapter = MagicMock()
+        mock_adapter.list_tags.return_value = ["3.18", "latest"]
+        mock_factory.return_value = mock_adapter
 
         result = ImageProvider.test_connection(image="alpine:3.18")
 
         assert result.is_connected is True
+        mock_adapter.list_tags.assert_called_once_with("library/alpine")
 
-    @patch("subprocess.run")
-    def test_test_connection_auth_failure(self, mock_subprocess):
-        """Test 401 error returns auth failure."""
-        mock_subprocess.return_value = MagicMock(
-            returncode=1, stderr="401 unauthorized"
+    @patch("prowler.providers.image.image_provider.create_registry_adapter")
+    def test_test_connection_auth_failure(self, mock_factory):
+        """Test registry auth error returns auth failure."""
+        mock_factory.return_value = MagicMock(
+            list_tags=MagicMock(side_effect=ImageRegistryAuthError(file=__file__))
         )
 
         result = ImageProvider.test_connection(image="private/image:latest")
@@ -309,10 +313,12 @@ class TestImageProvider:
         assert result.is_connected is False
         assert "Authentication failed" in result.error
 
-    @patch("subprocess.run")
-    def test_test_connection_not_found(self, mock_subprocess):
-        """Test 404 error returns not found."""
-        mock_subprocess.return_value = MagicMock(returncode=1, stderr="404 not found")
+    @patch("prowler.providers.image.image_provider.create_registry_adapter")
+    def test_test_connection_not_found(self, mock_factory):
+        """Test tag not found returns not found error."""
+        mock_adapter = MagicMock()
+        mock_adapter.list_tags.return_value = ["v1", "v2"]
+        mock_factory.return_value = mock_adapter
 
         result = ImageProvider.test_connection(image="nonexistent/image:latest")
 
@@ -578,10 +584,12 @@ class TestImageProviderRegistryAuth:
         assert env["TRIVY_USERNAME"] == "myuser"
         assert env["TRIVY_PASSWORD"] == "mypass"
 
-    @patch("subprocess.run")
-    def test_test_connection_with_basic_auth(self, mock_subprocess):
-        """Test test_connection uses Trivy native auth with TRIVY_USERNAME/PASSWORD env vars."""
-        mock_subprocess.return_value = MagicMock(returncode=0, stderr="")
+    @patch("prowler.providers.image.image_provider.create_registry_adapter")
+    def test_test_connection_with_basic_auth(self, mock_factory):
+        """Test test_connection passes credentials to the registry adapter."""
+        mock_adapter = MagicMock()
+        mock_adapter.list_tags.return_value = ["v1"]
+        mock_factory.return_value = mock_adapter
 
         result = ImageProvider.test_connection(
             image="private.registry.io/myapp:v1",
@@ -590,17 +598,19 @@ class TestImageProviderRegistryAuth:
         )
 
         assert result.is_connected is True
-        assert mock_subprocess.call_count == 1
-        trivy_call = mock_subprocess.call_args
-        assert trivy_call.args[0][0] == "trivy"
-        env = trivy_call.kwargs.get("env") or trivy_call[1].get("env")
-        assert env["TRIVY_USERNAME"] == "myuser"
-        assert env["TRIVY_PASSWORD"] == "mypass"
+        mock_factory.assert_called_once_with(
+            registry_url="private.registry.io",
+            username="myuser",
+            password="mypass",
+            token=None,
+        )
 
-    @patch("subprocess.run")
-    def test_test_connection_with_token(self, mock_subprocess):
-        """Test test_connection passes token via env."""
-        mock_subprocess.return_value = MagicMock(returncode=0, stderr="")
+    @patch("prowler.providers.image.image_provider.create_registry_adapter")
+    def test_test_connection_with_token(self, mock_factory):
+        """Test test_connection passes token to the registry adapter."""
+        mock_adapter = MagicMock()
+        mock_adapter.list_tags.return_value = ["v1"]
+        mock_factory.return_value = mock_adapter
 
         result = ImageProvider.test_connection(
             image="private.registry.io/myapp:v1",
@@ -608,9 +618,12 @@ class TestImageProviderRegistryAuth:
         )
 
         assert result.is_connected is True
-        call_kwargs = mock_subprocess.call_args
-        env = call_kwargs.kwargs.get("env") or call_kwargs[1].get("env")
-        assert env["TRIVY_REGISTRY_TOKEN"] == "my-token"
+        mock_factory.assert_called_once_with(
+            registry_url="private.registry.io",
+            username=None,
+            password=None,
+            token="my-token",
+        )
 
     def test_print_credentials_shows_auth_method(self):
         """Test that print_credentials outputs the auth method."""