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()
This commit is contained in:
Andoni A. committed 2026-02-19 16:27:04 +01:00
1 parent 8197418f71
commit 0caaddfb4a
4 files changed
+101 -17

No files matched your search

@@ -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",
+8 -13
View File
@@ -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,
+19 -1
View File
@@ -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
+72 -2
View File
@@ -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(