mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-23 04:21:52 +00:00
fix(image): route bare registry hostnames to OCI catalog in test_connection
The API passes provider.uid (a bare hostname) to test_connection(), which misclassified it as a Docker Hub image reference and failed auth against Docker Hub instead of the target registry. Add _is_registry_url() to detect bare hostnames (dots in host, no slash) and _test_registry_connection() to test connectivity via the OCI catalog endpoint. Also fix the UI placeholder to show a valid example without https://.
This commit is contained in:
@@ -871,6 +871,9 @@ class ImageProvider(Provider):
|
||||
"""
|
||||
Test connection to container registry by attempting to inspect an image.
|
||||
|
||||
For bare registry hostnames (e.g. ECR URLs passed by the API as provider_uid),
|
||||
uses the OCI catalog endpoint instead of trivy image.
|
||||
|
||||
Args:
|
||||
image: Container image to test
|
||||
raise_on_exception: Whether to raise exceptions
|
||||
@@ -889,6 +892,16 @@ class ImageProvider(Provider):
|
||||
if not image:
|
||||
return Connection(is_connected=False, error="Image name is required")
|
||||
|
||||
# Registry URL (bare hostname) → test via OCI catalog
|
||||
if ImageProvider._is_registry_url(image):
|
||||
return ImageProvider._test_registry_connection(
|
||||
registry_url=image,
|
||||
registry_username=registry_username,
|
||||
registry_password=registry_password,
|
||||
registry_token=registry_token,
|
||||
)
|
||||
|
||||
# Image reference → test via trivy
|
||||
# Build env with registry credentials
|
||||
env = dict(os.environ)
|
||||
if registry_username and registry_password:
|
||||
@@ -949,3 +962,37 @@ class ImageProvider(Provider):
|
||||
is_connected=False,
|
||||
error=f"Unexpected error: {str(error)}",
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _test_registry_connection(
|
||||
registry_url: str,
|
||||
registry_username: str | None = None,
|
||||
registry_password: str | None = None,
|
||||
registry_token: str | None = None,
|
||||
) -> "Connection":
|
||||
"""Test connection to a registry URL by listing repositories via OCI catalog."""
|
||||
try:
|
||||
adapter = create_registry_adapter(
|
||||
registry_url=registry_url,
|
||||
username=registry_username,
|
||||
password=registry_password,
|
||||
token=registry_token,
|
||||
)
|
||||
adapter.list_repositories()
|
||||
return Connection(is_connected=True)
|
||||
except Exception as error:
|
||||
error_str = str(error).lower()
|
||||
if "401" in error_str or "unauthorized" in error_str:
|
||||
return Connection(
|
||||
is_connected=False,
|
||||
error="Authentication failed. Check registry credentials.",
|
||||
)
|
||||
elif "404" in error_str or "not found" in error_str:
|
||||
return Connection(
|
||||
is_connected=False,
|
||||
error="Registry catalog not found.",
|
||||
)
|
||||
return Connection(
|
||||
is_connected=False,
|
||||
error=f"Failed to connect to registry: {str(error)[:200]}",
|
||||
)
|
||||
|
||||
@@ -616,6 +616,99 @@ class TestExtractRegistry:
|
||||
assert ImageProvider._extract_registry("nginx") is None
|
||||
|
||||
|
||||
class TestIsRegistryUrl:
|
||||
def test_bare_ecr_hostname(self):
|
||||
assert ImageProvider._is_registry_url(
|
||||
"714274078102.dkr.ecr.eu-west-1.amazonaws.com"
|
||||
)
|
||||
|
||||
def test_bare_hostname_with_port(self):
|
||||
assert ImageProvider._is_registry_url("myregistry.com:5000")
|
||||
|
||||
def test_bare_ghcr(self):
|
||||
assert ImageProvider._is_registry_url("ghcr.io")
|
||||
|
||||
def test_registry_with_namespace_only(self):
|
||||
"""Registry URL with a single path segment (no tag) is a registry URL."""
|
||||
assert ImageProvider._is_registry_url("ghcr.io/myorg")
|
||||
|
||||
def test_image_reference_not_registry(self):
|
||||
"""Full image reference with repo and tag is not a registry URL."""
|
||||
assert not ImageProvider._is_registry_url("ghcr.io/myorg/repo:tag")
|
||||
|
||||
def test_simple_image_name(self):
|
||||
assert not ImageProvider._is_registry_url("alpine:3.18")
|
||||
|
||||
def test_bare_image_no_tag(self):
|
||||
assert not ImageProvider._is_registry_url("nginx")
|
||||
|
||||
def test_dockerhub_namespace(self):
|
||||
assert not ImageProvider._is_registry_url("library/alpine")
|
||||
|
||||
|
||||
class TestTestRegistryConnection:
|
||||
@patch("prowler.providers.image.image_provider.create_registry_adapter")
|
||||
def test_registry_connection_success(self, mock_factory):
|
||||
"""Test that a bare hostname triggers registry catalog test."""
|
||||
mock_adapter = MagicMock()
|
||||
mock_adapter.list_repositories.return_value = ["repo1"]
|
||||
mock_factory.return_value = mock_adapter
|
||||
|
||||
result = ImageProvider.test_connection(
|
||||
image="714274078102.dkr.ecr.eu-west-1.amazonaws.com",
|
||||
registry_username="user",
|
||||
registry_password="pass",
|
||||
)
|
||||
|
||||
assert result.is_connected is True
|
||||
mock_factory.assert_called_once_with(
|
||||
registry_url="714274078102.dkr.ecr.eu-west-1.amazonaws.com",
|
||||
username="user",
|
||||
password="pass",
|
||||
token=None,
|
||||
)
|
||||
mock_adapter.list_repositories.assert_called_once()
|
||||
|
||||
@patch("prowler.providers.image.image_provider.create_registry_adapter")
|
||||
def test_registry_connection_auth_failure(self, mock_factory):
|
||||
"""Test that 401 from registry adapter returns auth failure."""
|
||||
mock_adapter = MagicMock()
|
||||
mock_adapter.list_repositories.side_effect = Exception("401 unauthorized")
|
||||
mock_factory.return_value = mock_adapter
|
||||
|
||||
result = ImageProvider.test_connection(
|
||||
image="714274078102.dkr.ecr.eu-west-1.amazonaws.com",
|
||||
)
|
||||
|
||||
assert result.is_connected is False
|
||||
assert "Authentication failed" in result.error
|
||||
|
||||
@patch("prowler.providers.image.image_provider.create_registry_adapter")
|
||||
def test_registry_connection_generic_error(self, mock_factory):
|
||||
"""Test that a generic error from registry adapter returns error message."""
|
||||
mock_adapter = MagicMock()
|
||||
mock_adapter.list_repositories.side_effect = Exception("connection refused")
|
||||
mock_factory.return_value = mock_adapter
|
||||
|
||||
result = ImageProvider.test_connection(
|
||||
image="myregistry.example.com",
|
||||
)
|
||||
|
||||
assert result.is_connected is False
|
||||
assert "Failed to connect to registry" in result.error
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_image_reference_still_uses_trivy(self, mock_subprocess):
|
||||
"""Test that a full image reference still uses trivy (not registry catalog)."""
|
||||
mock_subprocess.return_value = MagicMock(returncode=0, stderr="")
|
||||
|
||||
result = ImageProvider.test_connection(image="alpine:3.18")
|
||||
|
||||
assert result.is_connected is True
|
||||
assert mock_subprocess.call_count == 1
|
||||
assert mock_subprocess.call_args.args[0][0] == "trivy"
|
||||
|
||||
|
||||
class TestTrivyAuthIntegration:
|
||||
@patch("subprocess.run")
|
||||
def test_run_scan_passes_trivy_env_with_credentials(self, mock_subprocess):
|
||||
|
||||
@@ -60,7 +60,7 @@ const getProviderFieldDetails = (providerType?: ProviderType) => {
|
||||
case "image":
|
||||
return {
|
||||
label: "Registry URL",
|
||||
placeholder: "e.g. https://registry.example.com",
|
||||
placeholder: "e.g. 123456789012.dkr.ecr.us-east-1.amazonaws.com",
|
||||
};
|
||||
case "oraclecloud":
|
||||
return {
|
||||
|
||||
Reference in New Issue
Block a user