diff --git a/api/src/backend/api/tests/test_utils.py b/api/src/backend/api/tests/test_utils.py index 6a8920228f..06d0ab68a0 100644 --- a/api/src/backend/api/tests/test_utils.py +++ b/api/src/backend/api/tests/test_utils.py @@ -21,6 +21,7 @@ from prowler.providers.aws.aws_provider import AwsProvider from prowler.providers.aws.lib.security_hub.security_hub import SecurityHubConnection from prowler.providers.azure.azure_provider import AzureProvider from prowler.providers.cloudflare.cloudflare_provider import CloudflareProvider +from prowler.providers.common.models import Connection from prowler.providers.gcp.gcp_provider import GcpProvider from prowler.providers.github.github_provider import GithubProvider from prowler.providers.iac.iac_provider import IacProvider @@ -190,48 +191,45 @@ class TestProwlerProviderConnectionTest: assert isinstance(connection.error, Provider.secret.RelatedObjectDoesNotExist) assert str(connection.error) == "Provider has no secret." - @patch( - "prowler.providers.image.lib.registry.factory.create_registry_adapter", - ) @patch("api.utils.return_prowler_provider") def test_prowler_provider_connection_test_image_success( - self, mock_return_prowler_provider, mock_create_adapter + self, mock_return_prowler_provider ): provider = MagicMock() provider.provider = Provider.ProviderChoices.IMAGE.value provider.uid = "ghcr.io" provider.secret.secret = {"registry_token": "tok"} - mock_adapter = MagicMock() - mock_create_adapter.return_value = mock_adapter + prowler_provider = MagicMock() + prowler_provider.test_connection.return_value = Connection(is_connected=True) + mock_return_prowler_provider.return_value = prowler_provider connection = prowler_provider_connection_test(provider) assert connection.is_connected is True assert connection.error is None - mock_create_adapter.assert_called_once_with( - registry_url="ghcr.io", - username=None, - password=None, - token="tok", + prowler_provider.test_connection.assert_called_once_with( + registry="ghcr.io", + registry_username=None, + registry_password=None, + registry_token="tok", + raise_on_exception=False, ) - mock_adapter.list_repositories.assert_called_once() - @patch( - "prowler.providers.image.lib.registry.factory.create_registry_adapter", - ) @patch("api.utils.return_prowler_provider") def test_prowler_provider_connection_test_image_failure( - self, mock_return_prowler_provider, mock_create_adapter + self, mock_return_prowler_provider ): provider = MagicMock() provider.provider = Provider.ProviderChoices.IMAGE.value provider.uid = "ghcr.io" provider.secret.secret = {"registry_token": "bad-token"} - mock_adapter = MagicMock() - mock_adapter.list_repositories.side_effect = Exception("401 Unauthorized") - mock_create_adapter.return_value = mock_adapter + prowler_provider = MagicMock() + prowler_provider.test_connection.return_value = Connection( + is_connected=False, error="401 Unauthorized" + ) + mock_return_prowler_provider.return_value = prowler_provider connection = prowler_provider_connection_test(provider) diff --git a/api/src/backend/api/utils.py b/api/src/backend/api/utils.py index b82647d8c6..ec344ade16 100644 --- a/api/src/backend/api/utils.py +++ b/api/src/backend/api/utils.py @@ -323,19 +323,13 @@ 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) - except Exception as e: - return Connection(is_connected=False, error=str(e)) + return prowler_provider.test_connection( + registry=provider.uid, + registry_username=prowler_provider_kwargs.get("registry_username"), + registry_password=prowler_provider_kwargs.get("registry_password"), + registry_token=prowler_provider_kwargs.get("registry_token"), + raise_on_exception=False, + ) 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..1de03a7052 100644 --- a/prowler/providers/image/image_provider.py +++ b/prowler/providers/image/image_provider.py @@ -896,17 +896,23 @@ class ImageProvider(Provider): image: str | None = None, raise_on_exception: bool = True, provider_id: str | None = None, + registry: str | None = None, registry_username: str | None = None, registry_password: str | None = None, registry_token: str | None = None, ) -> "Connection": """ - Test connection to container registry by attempting to inspect an image. + Test connection to a container registry or individual image. + + When ``registry`` is provided, validates registry-level connectivity + by listing repositories through the registry adapter. Otherwise + falls back to testing access to an individual ``image`` via Trivy. Args: - image: Container image to test + image: Container image to test (individual image mode) raise_on_exception: Whether to raise exceptions provider_id: Fallback for image name + registry: Registry URL for registry-level connectivity test registry_username: Registry username for basic auth registry_password: Registry password for basic auth registry_token: Registry token for token-based auth @@ -915,6 +921,22 @@ class ImageProvider(Provider): Connection: Connection object with success status """ try: + # Registry-level connectivity test (used by API) + if registry: + from prowler.providers.image.lib.registry.factory import ( + create_registry_adapter, + ) + + adapter = create_registry_adapter( + registry_url=registry, + username=registry_username, + password=registry_password, + token=registry_token, + ) + adapter.list_repositories() + return Connection(is_connected=True) + + # Individual image test (existing behaviour) if provider_id and not image: image = provider_id