From 0caaddfb4a4fcbbc94234e0b163e41de93812b44 Mon Sep 17 00:00:00 2001 From: "Andoni A." <14891798+andoniaf@users.noreply.github.com> Date: Thu, 19 Feb 2026 16:27:04 +0100 Subject: [PATCH] fix(image): resolve migration conflict and refactor API connection test - Renumber image migration from 0076 to 0081 to avoid conflict with openstack - Add registry parameter to ImageProvider.test_connection() for registry-level testing - Refactor API connection test to delegate to test_connection() like other providers - Add tests for registry path in test_connection() --- ...age_provider.py => 0081_image_provider.py} | 3 +- api/src/backend/api/utils.py | 21 ++---- prowler/providers/image/image_provider.py | 20 ++++- tests/providers/image/image_provider_test.py | 74 ++++++++++++++++++- 4 files changed, 101 insertions(+), 17 deletions(-) rename api/src/backend/api/migrations/{0076_image_provider.py => 0081_image_provider.py} (91%) diff --git a/api/src/backend/api/migrations/0076_image_provider.py b/api/src/backend/api/migrations/0081_image_provider.py similarity index 91% rename from api/src/backend/api/migrations/0076_image_provider.py rename to api/src/backend/api/migrations/0081_image_provider.py index 234b10ea37..06db89991f 100644 --- a/api/src/backend/api/migrations/0076_image_provider.py +++ b/api/src/backend/api/migrations/0081_image_provider.py @@ -7,7 +7,7 @@ import api.db_utils class Migration(migrations.Migration): dependencies = [ - ("api", "0075_cloudflare_provider"), + ("api", "0080_backfill_attack_paths_graph_data_ready"), ] operations = [ @@ -27,6 +27,7 @@ class Migration(migrations.Migration): ("oraclecloud", "Oracle Cloud Infrastructure"), ("alibabacloud", "Alibaba Cloud"), ("cloudflare", "Cloudflare"), + ("openstack", "OpenStack"), ("image", "Image"), ], default="aws", diff --git a/api/src/backend/api/utils.py b/api/src/backend/api/utils.py index f80fc17650..a04a6b1828 100644 --- a/api/src/backend/api/utils.py +++ b/api/src/backend/api/utils.py @@ -324,19 +324,14 @@ def prowler_provider_connection_test(provider: Provider) -> Connection: } return prowler_provider.test_connection(**openstack_kwargs) elif provider.provider == Provider.ProviderChoices.IMAGE.value: - from prowler.providers.image.lib.registry.factory import create_registry_adapter - - try: - adapter = create_registry_adapter( - registry_url=provider.uid, - username=prowler_provider_kwargs.get("registry_username"), - password=prowler_provider_kwargs.get("registry_password"), - token=prowler_provider_kwargs.get("registry_token"), - ) - adapter.list_repositories() - return Connection(is_connected=True, error=None) - except Exception as e: - return Connection(is_connected=False, error=str(e)) + image_kwargs = { + "registry": provider.uid, + "raise_on_exception": False, + } + for key in ("registry_username", "registry_password", "registry_token"): + if key in prowler_provider_kwargs: + image_kwargs[key] = prowler_provider_kwargs[key] + return prowler_provider.test_connection(**image_kwargs) else: return prowler_provider.test_connection( **prowler_provider_kwargs, diff --git a/prowler/providers/image/image_provider.py b/prowler/providers/image/image_provider.py index be16580ba1..f6d63b4be2 100644 --- a/prowler/providers/image/image_provider.py +++ b/prowler/providers/image/image_provider.py @@ -899,9 +899,14 @@ class ImageProvider(Provider): registry_username: str | None = None, registry_password: str | None = None, registry_token: str | None = None, + registry: str | None = None, ) -> "Connection": """ - Test connection to container registry by attempting to inspect an image. + Test connection to a container registry or image. + + When ``registry`` is provided (and no ``image``), tests registry-level + reachability via the OCI catalog endpoint (list repositories). + Otherwise falls back to the Trivy-based single-image test. Args: image: Container image to test @@ -910,11 +915,24 @@ class ImageProvider(Provider): registry_username: Registry username for basic auth registry_password: Registry password for basic auth registry_token: Registry token for token-based auth + registry: Registry URL for registry-level connectivity test Returns: Connection: Connection object with success status """ try: + # Registry-level test: list repositories to verify reachability + if registry and not image: + adapter = create_registry_adapter( + registry_url=registry, + username=registry_username, + password=registry_password, + token=registry_token, + ) + adapter.list_repositories() + return Connection(is_connected=True) + + # Image-level test via Trivy if provider_id and not image: image = provider_id diff --git a/tests/providers/image/image_provider_test.py b/tests/providers/image/image_provider_test.py index 74ce4fd47e..758031b872 100644 --- a/tests/providers/image/image_provider_test.py +++ b/tests/providers/image/image_provider_test.py @@ -7,8 +7,6 @@ import pytest from prowler.lib.check.models import CheckReportImage from prowler.providers.image.exceptions.exceptions import ( - ImageDockerLoginError, - ImageDockerNotFoundError, ImageInvalidConfigScannerError, ImageInvalidNameError, ImageInvalidScannerError, @@ -805,6 +803,78 @@ class TestImageProviderRegistryAuth: env = call_kwargs.kwargs.get("env") or call_kwargs[1].get("env") assert env["TRIVY_REGISTRY_TOKEN"] == "my-token" + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_test_connection_registry_success(self, mock_factory): + """Test registry-level connection test via list_repositories.""" + mock_adapter = MagicMock() + mock_adapter.list_repositories.return_value = ["repo1", "repo2"] + mock_factory.return_value = mock_adapter + + result = ImageProvider.test_connection( + registry="https://myregistry.io", + registry_username="user", + registry_password="pass", + ) + + assert result.is_connected is True + mock_factory.assert_called_once_with( + registry_url="https://myregistry.io", + username="user", + password="pass", + token=None, + ) + mock_adapter.list_repositories.assert_called_once() + + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_test_connection_registry_failure(self, mock_factory): + """Test registry-level connection test failure.""" + mock_adapter = MagicMock() + mock_adapter.list_repositories.side_effect = Exception("401 Unauthorized") + mock_factory.return_value = mock_adapter + + result = ImageProvider.test_connection( + registry="https://myregistry.io", + raise_on_exception=False, + ) + + assert result.is_connected is False + assert "401 Unauthorized" in result.error + + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_test_connection_registry_with_token(self, mock_factory): + """Test registry-level connection test with token auth.""" + mock_adapter = MagicMock() + mock_adapter.list_repositories.return_value = ["repo1"] + mock_factory.return_value = mock_adapter + + result = ImageProvider.test_connection( + registry="https://ghcr.io", + registry_token="my-token", + ) + + assert result.is_connected is True + mock_factory.assert_called_once_with( + registry_url="https://ghcr.io", + username=None, + password=None, + token="my-token", + ) + + @patch("subprocess.run") + def test_test_connection_image_takes_precedence_over_registry( + self, mock_subprocess + ): + """When both image and registry are provided, image-level Trivy test is used.""" + mock_subprocess.return_value = MagicMock(returncode=0, stderr="") + + result = ImageProvider.test_connection( + image="alpine:3.18", + registry="https://myregistry.io", + ) + + assert result.is_connected is True + mock_subprocess.assert_called_once() + def test_print_credentials_shows_auth_method(self): """Test that print_credentials outputs the auth method.""" provider = _make_provider(