From 57ed5a701d19a1162d61abbddac46a284f17757f Mon Sep 17 00:00:00 2001 From: "Andoni A." <14891798+andoniaf@users.noreply.github.com> Date: Thu, 19 Feb 2026 18:24:37 +0100 Subject: [PATCH] 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 | 104 ++++++++++--------- tests/providers/image/image_provider_test.py | 71 +++++++------ 2 files changed, 96 insertions(+), 79 deletions(-) diff --git a/prowler/providers/image/image_provider.py b/prowler/providers/image/image_provider.py index da5fac5604..e096595f14 100644 --- a/prowler/providers/image/image_provider.py +++ b/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: diff --git a/tests/providers/image/image_provider_test.py b/tests/providers/image/image_provider_test.py index fc36b9e617..8ce9c41a83 100644 --- a/tests/providers/image/image_provider_test.py +++ b/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."""