test(image): add comprehensive registry scan mode unit tests

69 tests covering:
- Factory auto-detection (OCI, Docker Hub, ECR URL patterns)
- OCI adapter (auth flows, pagination, retry, error handling)
- Docker Hub adapter (namespace extraction, Hub API, token exchange)
- Argument validation (registry flags, Docker Hub namespace)
- Provider integration (enumeration, filtering, max-images, dedup)
This commit is contained in:
Andoni A.
2026-02-16 15:42:02 +01:00
parent ace36fe76e
commit aa284b8b24
5 changed files with 574 additions and 0 deletions
@@ -0,0 +1,79 @@
from argparse import Namespace
import pytest
from prowler.providers.image.lib.arguments.arguments import validate_arguments
class TestValidateArguments:
def test_no_source_fails(self):
args = Namespace(images=[], image_list_file=None, registry=None, image_filter=None, tag_filter=None, max_images=0, registry_insecure=False)
ok, msg = validate_arguments(args)
assert not ok
assert "--image" in msg
def test_image_only_passes(self):
args = Namespace(images=["nginx:latest"], image_list_file=None, registry=None, image_filter=None, tag_filter=None, max_images=0, registry_insecure=False)
ok, _ = validate_arguments(args)
assert ok
def test_image_list_only_passes(self):
args = Namespace(images=[], image_list_file="images.txt", registry=None, image_filter=None, tag_filter=None, max_images=0, registry_insecure=False)
ok, _ = validate_arguments(args)
assert ok
def test_registry_only_passes(self):
args = Namespace(images=[], image_list_file=None, registry="myregistry.io", image_filter=None, tag_filter=None, max_images=0, registry_insecure=False)
ok, _ = validate_arguments(args)
assert ok
def test_image_filter_without_registry_fails(self):
args = Namespace(images=["nginx:latest"], image_list_file=None, registry=None, image_filter="^prod", tag_filter=None, max_images=0, registry_insecure=False)
ok, msg = validate_arguments(args)
assert not ok
assert "--image-filter requires --registry" in msg
def test_tag_filter_without_registry_fails(self):
args = Namespace(images=["nginx:latest"], image_list_file=None, registry=None, image_filter=None, tag_filter="^v", max_images=0, registry_insecure=False)
ok, msg = validate_arguments(args)
assert not ok
assert "--tag-filter requires --registry" in msg
def test_max_images_without_registry_fails(self):
args = Namespace(images=["nginx:latest"], image_list_file=None, registry=None, image_filter=None, tag_filter=None, max_images=50, registry_insecure=False)
ok, msg = validate_arguments(args)
assert not ok
assert "--max-images requires --registry" in msg
def test_registry_insecure_without_registry_fails(self):
args = Namespace(images=[], image_list_file="i.txt", registry=None, image_filter=None, tag_filter=None, max_images=0, registry_insecure=True)
ok, msg = validate_arguments(args)
assert not ok
assert "--registry-insecure requires --registry" in msg
def test_docker_hub_no_namespace_fails(self):
args = Namespace(images=[], image_list_file=None, registry="docker.io", image_filter=None, tag_filter=None, max_images=0, registry_insecure=False)
ok, msg = validate_arguments(args)
assert not ok
assert "namespace" in msg.lower()
def test_docker_hub_with_namespace_passes(self):
args = Namespace(images=[], image_list_file=None, registry="docker.io/myorg", image_filter=None, tag_filter=None, max_images=0, registry_insecure=False)
ok, _ = validate_arguments(args)
assert ok
def test_docker_hub_https_no_namespace_fails(self):
args = Namespace(images=[], image_list_file=None, registry="https://docker.io", image_filter=None, tag_filter=None, max_images=0, registry_insecure=False)
ok, msg = validate_arguments(args)
assert not ok
assert "namespace" in msg.lower()
def test_registry_with_filters_passes(self):
args = Namespace(images=[], image_list_file=None, registry="myregistry.io", image_filter="^prod", tag_filter="^v", max_images=100, registry_insecure=True)
ok, _ = validate_arguments(args)
assert ok
def test_combined_registry_and_image_passes(self):
args = Namespace(images=["nginx:latest"], image_list_file=None, registry="myregistry.io", image_filter=None, tag_filter=None, max_images=0, registry_insecure=False)
ok, _ = validate_arguments(args)
assert ok
@@ -0,0 +1,117 @@
from unittest.mock import MagicMock, patch
import pytest
import requests
from prowler.providers.image.exceptions.exceptions import (
ImageRegistryAuthError,
ImageRegistryNetworkError,
)
from prowler.providers.image.lib.registry.dockerhub_adapter import DockerHubAdapter
class TestDockerHubAdapterInit:
def test_extract_namespace_simple(self):
assert DockerHubAdapter._extract_namespace("docker.io/myorg") == "myorg"
def test_extract_namespace_https(self):
assert DockerHubAdapter._extract_namespace("https://docker.io/myorg") == "myorg"
def test_extract_namespace_registry1(self):
assert DockerHubAdapter._extract_namespace("registry-1.docker.io/myorg") == "myorg"
def test_extract_namespace_empty(self):
assert DockerHubAdapter._extract_namespace("docker.io") == ""
def test_extract_namespace_with_slash(self):
assert DockerHubAdapter._extract_namespace("docker.io/myorg/") == "myorg"
class TestDockerHubListRepositories:
@patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.request")
@patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.post")
def test_list_repos(self, mock_post, mock_request):
# Hub login
login_resp = MagicMock(status_code=200)
login_resp.json.return_value = {"token": "jwt"}
mock_post.return_value = login_resp
# Repo listing
repos_resp = MagicMock(status_code=200)
repos_resp.json.return_value = {
"results": [{"name": "app1"}, {"name": "app2"}],
"next": None,
}
mock_request.return_value = repos_resp
adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p")
repos = adapter.list_repositories()
assert repos == ["myorg/app1", "myorg/app2"]
def test_list_repos_no_namespace_raises(self):
adapter = DockerHubAdapter("docker.io")
with pytest.raises(ImageRegistryAuthError, match="namespace"):
adapter.list_repositories()
class TestDockerHubListTags:
@patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.request")
@patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.get")
def test_list_tags(self, mock_get, mock_request):
# Token exchange
token_resp = MagicMock(status_code=200)
token_resp.json.return_value = {"token": "registry-token"}
mock_get.return_value = token_resp
# Tag listing
tags_resp = MagicMock(status_code=200, headers={})
tags_resp.json.return_value = {"tags": ["latest", "v1.0"]}
mock_request.return_value = tags_resp
adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p")
tags = adapter.list_tags("myorg/myapp")
assert tags == ["latest", "v1.0"]
@patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.request")
@patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.get")
def test_list_tags_auth_failure(self, mock_get, mock_request):
token_resp = MagicMock(status_code=200)
token_resp.json.return_value = {"token": "tok"}
mock_get.return_value = token_resp
tags_resp = MagicMock(status_code=401)
mock_request.return_value = tags_resp
adapter = DockerHubAdapter("docker.io/myorg")
with pytest.raises(ImageRegistryAuthError):
adapter.list_tags("myorg/myapp")
class TestDockerHubLogin:
@patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.post")
def test_login_failure(self, mock_post):
resp = MagicMock(status_code=401)
mock_post.return_value = resp
adapter = DockerHubAdapter("docker.io/myorg", username="bad", password="creds")
with pytest.raises(ImageRegistryAuthError, match="login failed"):
adapter._hub_login()
def test_login_skipped_without_credentials(self):
adapter = DockerHubAdapter("docker.io/myorg")
adapter._hub_login() # Should not raise
assert adapter._hub_jwt is None
class TestDockerHubRetry:
@patch("prowler.providers.image.lib.registry.dockerhub_adapter.time.sleep")
@patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.request")
def test_retry_on_429(self, mock_request, mock_sleep):
resp_429 = MagicMock(status_code=429)
resp_200 = MagicMock(status_code=200)
mock_request.side_effect = [resp_429, resp_200]
adapter = DockerHubAdapter("docker.io/myorg")
result = adapter._request_with_retry("GET", "https://hub.docker.com/v2/namespaces/myorg/repositories")
assert result.status_code == 200
@patch("prowler.providers.image.lib.registry.dockerhub_adapter.time.sleep")
@patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.request")
def test_connection_error_retries(self, mock_request, mock_sleep):
mock_request.side_effect = requests.exceptions.ConnectionError("fail")
adapter = DockerHubAdapter("docker.io/myorg")
with pytest.raises(ImageRegistryNetworkError):
adapter._request_with_retry("GET", "https://hub.docker.com")
assert mock_request.call_count == 3
@@ -0,0 +1,63 @@
from unittest.mock import patch
import pytest
from prowler.providers.image.lib.registry.dockerhub_adapter import DockerHubAdapter
from prowler.providers.image.lib.registry.factory import (
create_registry_adapter,
detect_registry_type,
)
from prowler.providers.image.lib.registry.oci_adapter import OciRegistryAdapter
class TestDetectRegistryType:
def test_generic_oci(self):
assert detect_registry_type("myregistry.io") == "oci"
def test_generic_oci_with_https(self):
assert detect_registry_type("https://myregistry.io") == "oci"
def test_docker_hub(self):
assert detect_registry_type("docker.io/myorg") == "dockerhub"
def test_docker_hub_registry1(self):
assert detect_registry_type("registry-1.docker.io/myorg") == "dockerhub"
def test_docker_hub_with_https(self):
assert detect_registry_type("https://docker.io/myorg") == "dockerhub"
def test_ecr(self):
assert detect_registry_type("123456789.dkr.ecr.us-east-1.amazonaws.com") == "ecr"
def test_ecr_with_https(self):
assert detect_registry_type("https://123456789.dkr.ecr.us-east-1.amazonaws.com") == "ecr"
def test_harbor(self):
assert detect_registry_type("harbor.example.com") == "oci"
class TestCreateRegistryAdapter:
def test_docker_hub_returns_dockerhub_adapter(self):
adapter = create_registry_adapter("docker.io/myorg")
assert isinstance(adapter, DockerHubAdapter)
def test_oci_returns_oci_adapter(self):
adapter = create_registry_adapter("myregistry.io")
assert isinstance(adapter, OciRegistryAdapter)
def test_ecr_returns_oci_adapter(self):
adapter = create_registry_adapter("123456789.dkr.ecr.us-east-1.amazonaws.com")
assert isinstance(adapter, OciRegistryAdapter)
def test_passes_credentials(self):
adapter = create_registry_adapter(
"myregistry.io", username="user", password="pass", token="tok", verify_ssl=False
)
assert adapter.username == "user"
assert adapter.password == "pass"
assert adapter.token == "tok"
assert adapter.verify_ssl is False
def test_registry_1_docker_io(self):
adapter = create_registry_adapter("registry-1.docker.io/myorg")
assert isinstance(adapter, DockerHubAdapter)
@@ -0,0 +1,166 @@
from unittest.mock import MagicMock, patch
import pytest
import requests
from prowler.providers.image.exceptions.exceptions import (
ImageRegistryAuthError,
ImageRegistryCatalogError,
ImageRegistryNetworkError,
)
from prowler.providers.image.lib.registry.oci_adapter import OciRegistryAdapter
class TestOciAdapterInit:
def test_normalise_url_adds_https(self):
adapter = OciRegistryAdapter("myregistry.io")
assert adapter._base_url == "https://myregistry.io"
def test_normalise_url_keeps_http(self):
adapter = OciRegistryAdapter("http://myregistry.io")
assert adapter._base_url == "http://myregistry.io"
def test_normalise_url_strips_trailing_slash(self):
adapter = OciRegistryAdapter("https://myregistry.io/")
assert adapter._base_url == "https://myregistry.io"
def test_stores_credentials(self):
adapter = OciRegistryAdapter("reg.io", username="u", password="p", token="t", verify_ssl=False)
assert adapter.username == "u"
assert adapter.password == "p"
assert adapter.token == "t"
assert adapter.verify_ssl is False
class TestOciAdapterAuth:
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
def test_ensure_auth_with_token(self, mock_request):
adapter = OciRegistryAdapter("reg.io", token="my-token")
adapter._ensure_auth()
assert adapter._bearer_token == "my-token"
mock_request.assert_not_called()
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
def test_ensure_auth_anonymous_ok(self, mock_request):
resp = MagicMock(status_code=200)
mock_request.return_value = resp
adapter = OciRegistryAdapter("reg.io")
adapter._ensure_auth()
assert adapter._bearer_token is None
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
def test_ensure_auth_bearer_challenge(self, mock_request):
ping_resp = MagicMock(status_code=401, headers={"Www-Authenticate": 'Bearer realm="https://auth.example.com/token",service="registry"'})
token_resp = MagicMock(status_code=200)
token_resp.json.return_value = {"token": "bearer-tok"}
mock_request.side_effect = [ping_resp, token_resp]
adapter = OciRegistryAdapter("reg.io", username="u", password="p")
adapter._ensure_auth()
assert adapter._bearer_token == "bearer-tok"
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
def test_ensure_auth_403_raises(self, mock_request):
resp = MagicMock(status_code=403)
mock_request.return_value = resp
adapter = OciRegistryAdapter("reg.io")
with pytest.raises(ImageRegistryAuthError):
adapter._ensure_auth()
class TestOciAdapterListRepositories:
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
def test_list_repos_single_page(self, mock_request):
ping_resp = MagicMock(status_code=200)
catalog_resp = MagicMock(status_code=200, headers={})
catalog_resp.json.return_value = {"repositories": ["app/frontend", "app/backend"]}
mock_request.side_effect = [ping_resp, catalog_resp]
adapter = OciRegistryAdapter("reg.io")
repos = adapter.list_repositories()
assert repos == ["app/frontend", "app/backend"]
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
def test_list_repos_paginated(self, mock_request):
ping_resp = MagicMock(status_code=200)
page1_resp = MagicMock(status_code=200, headers={"Link": '<https://reg.io/v2/_catalog?n=200&last=b>; rel="next"'})
page1_resp.json.return_value = {"repositories": ["a"]}
page2_resp = MagicMock(status_code=200, headers={})
page2_resp.json.return_value = {"repositories": ["b"]}
mock_request.side_effect = [ping_resp, page1_resp, page2_resp]
adapter = OciRegistryAdapter("reg.io")
repos = adapter.list_repositories()
assert repos == ["a", "b"]
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
def test_list_repos_404_raises(self, mock_request):
ping_resp = MagicMock(status_code=200)
catalog_resp = MagicMock(status_code=404)
mock_request.side_effect = [ping_resp, catalog_resp]
adapter = OciRegistryAdapter("reg.io")
with pytest.raises(ImageRegistryCatalogError):
adapter.list_repositories()
class TestOciAdapterListTags:
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
def test_list_tags(self, mock_request):
ping_resp = MagicMock(status_code=200)
tags_resp = MagicMock(status_code=200, headers={})
tags_resp.json.return_value = {"tags": ["latest", "v1.0"]}
mock_request.side_effect = [ping_resp, tags_resp]
adapter = OciRegistryAdapter("reg.io")
tags = adapter.list_tags("myapp")
assert tags == ["latest", "v1.0"]
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
def test_list_tags_null_tags(self, mock_request):
ping_resp = MagicMock(status_code=200)
tags_resp = MagicMock(status_code=200, headers={})
tags_resp.json.return_value = {"tags": None}
mock_request.side_effect = [ping_resp, tags_resp]
adapter = OciRegistryAdapter("reg.io")
tags = adapter.list_tags("myapp")
assert tags == []
class TestOciAdapterRetry:
@patch("prowler.providers.image.lib.registry.oci_adapter.time.sleep")
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
def test_retry_on_429(self, mock_request, mock_sleep):
resp_429 = MagicMock(status_code=429)
resp_200 = MagicMock(status_code=200)
mock_request.side_effect = [resp_429, resp_200]
adapter = OciRegistryAdapter("reg.io")
result = adapter._request_with_retry("GET", "https://reg.io/v2/")
assert result.status_code == 200
mock_sleep.assert_called_once()
@patch("prowler.providers.image.lib.registry.oci_adapter.time.sleep")
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
def test_connection_error_retries(self, mock_request, mock_sleep):
mock_request.side_effect = requests.exceptions.ConnectionError("failed")
adapter = OciRegistryAdapter("reg.io")
with pytest.raises(ImageRegistryNetworkError):
adapter._request_with_retry("GET", "https://reg.io/v2/")
assert mock_request.call_count == 3
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
def test_timeout_raises_immediately(self, mock_request):
mock_request.side_effect = requests.exceptions.Timeout("timeout")
adapter = OciRegistryAdapter("reg.io")
with pytest.raises(ImageRegistryNetworkError):
adapter._request_with_retry("GET", "https://reg.io/v2/")
assert mock_request.call_count == 1
class TestOciAdapterNextPageUrl:
def test_no_link_header(self):
resp = MagicMock(headers={})
assert OciRegistryAdapter._next_page_url(resp) is None
def test_link_header_with_next(self):
resp = MagicMock(headers={"Link": '<https://reg.io/v2/_catalog?n=200&last=b>; rel="next"'})
assert OciRegistryAdapter._next_page_url(resp) == "https://reg.io/v2/_catalog?n=200&last=b"
def test_link_header_no_next(self):
resp = MagicMock(headers={"Link": '<https://reg.io/v2/_catalog?n=200>; rel="prev"'})
assert OciRegistryAdapter._next_page_url(resp) is None
@@ -0,0 +1,149 @@
import os
from argparse import Namespace
from unittest.mock import MagicMock, patch
import pytest
from prowler.providers.image.exceptions.exceptions import (
ImageInvalidFilterError,
ImageMaxImagesExceededError,
)
from prowler.providers.image.image_provider import ImageProvider
_CLEAN_ENV = {
"PATH": os.environ.get("PATH", ""),
"HOME": os.environ.get("HOME", ""),
}
def _build_provider(**overrides):
defaults = dict(
images=[],
registry="myregistry.io",
image_filter=None,
tag_filter=None,
max_images=0,
registry_insecure=False,
config_content={"image": {}},
)
defaults.update(overrides)
with patch.dict(os.environ, _CLEAN_ENV, clear=True):
return ImageProvider(**defaults)
class TestRegistryEnumeration:
@patch("prowler.providers.image.image_provider.create_registry_adapter")
def test_enumerate_oci_registry(self, mock_factory):
adapter = MagicMock()
adapter.list_repositories.return_value = ["app/frontend", "app/backend"]
adapter.list_tags.side_effect = [["latest", "v1.0"], ["latest"]]
mock_factory.return_value = adapter
provider = _build_provider()
assert "myregistry.io/app/frontend:latest" in provider.images
assert "myregistry.io/app/frontend:v1.0" in provider.images
assert "myregistry.io/app/backend:latest" in provider.images
assert len(provider.images) == 3
@patch("prowler.providers.image.image_provider.create_registry_adapter")
def test_image_filter(self, mock_factory):
adapter = MagicMock()
adapter.list_repositories.return_value = ["prod/app", "dev/app", "staging/app"]
adapter.list_tags.return_value = ["latest"]
mock_factory.return_value = adapter
provider = _build_provider(image_filter="^prod/")
assert len(provider.images) == 1
assert "myregistry.io/prod/app:latest" in provider.images
@patch("prowler.providers.image.image_provider.create_registry_adapter")
def test_tag_filter(self, mock_factory):
adapter = MagicMock()
adapter.list_repositories.return_value = ["myapp"]
adapter.list_tags.return_value = ["latest", "v1.0", "v2.0", "dev-abc123"]
mock_factory.return_value = adapter
provider = _build_provider(tag_filter=r"^v\d+\.\d+$")
assert len(provider.images) == 2
assert "myregistry.io/myapp:v1.0" in provider.images
assert "myregistry.io/myapp:v2.0" in provider.images
@patch("prowler.providers.image.image_provider.create_registry_adapter")
def test_combined_filters(self, mock_factory):
adapter = MagicMock()
adapter.list_repositories.return_value = ["prod/app", "dev/app"]
adapter.list_tags.return_value = ["latest", "v1.0"]
mock_factory.return_value = adapter
provider = _build_provider(image_filter="^prod/", tag_filter="^v")
assert len(provider.images) == 1
assert "myregistry.io/prod/app:v1.0" in provider.images
class TestMaxImages:
@patch("prowler.providers.image.image_provider.create_registry_adapter")
def test_max_images_exceeded(self, mock_factory):
adapter = MagicMock()
adapter.list_repositories.return_value = ["app1", "app2", "app3"]
adapter.list_tags.return_value = ["latest", "v1.0"]
mock_factory.return_value = adapter
with pytest.raises(ImageMaxImagesExceededError):
_build_provider(max_images=2)
@patch("prowler.providers.image.image_provider.create_registry_adapter")
def test_max_images_not_exceeded(self, mock_factory):
adapter = MagicMock()
adapter.list_repositories.return_value = ["app1"]
adapter.list_tags.return_value = ["latest"]
mock_factory.return_value = adapter
provider = _build_provider(max_images=10)
assert len(provider.images) == 1
class TestDeduplication:
@patch("prowler.providers.image.image_provider.create_registry_adapter")
def test_deduplication_with_explicit_images(self, mock_factory):
adapter = MagicMock()
adapter.list_repositories.return_value = ["myapp"]
adapter.list_tags.return_value = ["latest"]
mock_factory.return_value = adapter
provider = _build_provider(images=["myregistry.io/myapp:latest"])
assert provider.images.count("myregistry.io/myapp:latest") == 1
class TestInvalidFilters:
def test_invalid_image_filter_regex(self):
with pytest.raises(ImageInvalidFilterError):
_build_provider(image_filter="[invalid")
def test_invalid_tag_filter_regex(self):
with pytest.raises(ImageInvalidFilterError):
_build_provider(tag_filter="(unclosed")
class TestRegistryInsecure:
@patch("prowler.providers.image.image_provider.create_registry_adapter")
def test_insecure_passes_verify_false(self, mock_factory):
adapter = MagicMock()
adapter.list_repositories.return_value = ["app"]
adapter.list_tags.return_value = ["latest"]
mock_factory.return_value = adapter
_build_provider(registry_insecure=True)
mock_factory.assert_called_once()
call_kwargs = mock_factory.call_args[1]
assert call_kwargs["verify_ssl"] is False
class TestEmptyRegistry:
@patch("prowler.providers.image.image_provider.create_registry_adapter")
def test_empty_catalog_with_explicit_images(self, mock_factory):
adapter = MagicMock()
adapter.list_repositories.return_value = []
mock_factory.return_value = adapter
provider = _build_provider(images=["nginx:latest"])
assert provider.images == ["nginx:latest"]