test(image): add tests for SSRF, credential redaction, empty tokens, and DockerHub enumeration

- SSRF: reject file/ftp schemes, private/loopback/link-local IPs, accept public URLs
- Credential redaction: __getstate__ redacts password/token, properties still work
- Empty token guards: OCI bearer, DockerHub JWT, DockerHub registry token
- Narrow except: invalid UTF-8 base64 falls through correctly
- Fix namespace test to expect ImageRegistryCatalogError instead of AuthError
- Add registry_list_images=False to all Namespace objects missing it
- Add DockerHub enumeration integration test verifying repo:tag format
This commit is contained in:
Andoni A. committed 2026-02-17 11:40:31 +01:00
1 parent a692351f93
commit 15557b9f6c
4 files changed
+172 -1

No files matched your search

@@ -13,6 +13,7 @@ class TestValidateArguments:
tag_filter=None,
max_images=0,
registry_insecure=False,
registry_list_images=False,
)
ok, msg = validate_arguments(args)
assert not ok
@@ -27,6 +28,7 @@ class TestValidateArguments:
tag_filter=None,
max_images=0,
registry_insecure=False,
registry_list_images=False,
)
ok, _ = validate_arguments(args)
assert ok
@@ -40,6 +42,7 @@ class TestValidateArguments:
tag_filter=None,
max_images=0,
registry_insecure=False,
registry_list_images=False,
)
ok, _ = validate_arguments(args)
assert ok
@@ -53,6 +56,7 @@ class TestValidateArguments:
tag_filter=None,
max_images=0,
registry_insecure=False,
registry_list_images=False,
)
ok, _ = validate_arguments(args)
assert ok
@@ -66,6 +70,7 @@ class TestValidateArguments:
tag_filter=None,
max_images=0,
registry_insecure=False,
registry_list_images=False,
)
ok, msg = validate_arguments(args)
assert not ok
@@ -80,6 +85,7 @@ class TestValidateArguments:
tag_filter="^v",
max_images=0,
registry_insecure=False,
registry_list_images=False,
)
ok, msg = validate_arguments(args)
assert not ok
@@ -94,6 +100,7 @@ class TestValidateArguments:
tag_filter=None,
max_images=50,
registry_insecure=False,
registry_list_images=False,
)
ok, msg = validate_arguments(args)
assert not ok
@@ -108,6 +115,7 @@ class TestValidateArguments:
tag_filter=None,
max_images=0,
registry_insecure=True,
registry_list_images=False,
)
ok, msg = validate_arguments(args)
assert not ok
@@ -122,6 +130,7 @@ class TestValidateArguments:
tag_filter=None,
max_images=0,
registry_insecure=False,
registry_list_images=False,
)
ok, msg = validate_arguments(args)
assert not ok
@@ -136,6 +145,7 @@ class TestValidateArguments:
tag_filter=None,
max_images=0,
registry_insecure=False,
registry_list_images=False,
)
ok, _ = validate_arguments(args)
assert ok
@@ -149,6 +159,7 @@ class TestValidateArguments:
tag_filter=None,
max_images=0,
registry_insecure=False,
registry_list_images=False,
)
ok, msg = validate_arguments(args)
assert not ok
@@ -163,6 +174,7 @@ class TestValidateArguments:
tag_filter="^v",
max_images=100,
registry_insecure=True,
registry_list_images=False,
)
ok, _ = validate_arguments(args)
assert ok
@@ -205,6 +217,7 @@ class TestValidateArguments:
tag_filter=None,
max_images=0,
registry_insecure=False,
registry_list_images=False,
)
ok, _ = validate_arguments(args)
assert ok
@@ -5,6 +5,7 @@ import requests
from prowler.providers.image.exceptions.exceptions import (
ImageRegistryAuthError,
ImageRegistryCatalogError,
ImageRegistryNetworkError,
)
from prowler.providers.image.lib.registry.dockerhub_adapter import DockerHubAdapter
@@ -48,7 +49,7 @@ class TestDockerHubListRepositories:
def test_list_repos_no_namespace_raises(self):
adapter = DockerHubAdapter("docker.io")
with pytest.raises(ImageRegistryAuthError, match="namespace"):
with pytest.raises(ImageRegistryCatalogError, match="namespace"):
adapter.list_repositories()
@patch("prowler.providers.image.lib.registry.base.requests.request")
@@ -131,3 +132,32 @@ class TestDockerHubRetry:
with pytest.raises(ImageRegistryNetworkError):
adapter._request_with_retry("GET", "https://hub.docker.com")
assert mock_request.call_count == 3
class TestDockerHubEmptyTokens:
@patch("prowler.providers.image.lib.registry.base.requests.request")
def test_empty_hub_jwt_raises(self, mock_request):
resp = MagicMock(status_code=200)
resp.json.return_value = {"token": ""}
mock_request.return_value = resp
adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p")
with pytest.raises(ImageRegistryAuthError, match="empty JWT"):
adapter._hub_login()
@patch("prowler.providers.image.lib.registry.base.requests.request")
def test_none_hub_jwt_raises(self, mock_request):
resp = MagicMock(status_code=200)
resp.json.return_value = {}
mock_request.return_value = resp
adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p")
with pytest.raises(ImageRegistryAuthError, match="empty JWT"):
adapter._hub_login()
@patch("prowler.providers.image.lib.registry.base.requests.request")
def test_empty_registry_token_raises(self, mock_request):
resp = MagicMock(status_code=200)
resp.json.return_value = {"token": ""}
mock_request.return_value = resp
adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p")
with pytest.raises(ImageRegistryAuthError, match="empty token"):
adapter._get_registry_token("myorg/myapp")
@@ -305,3 +305,110 @@ class TestOciAdapterNextPageUrl:
headers={"Link": '<https://reg.io/v2/_catalog?n=200>; rel="prev"'}
)
assert OciRegistryAdapter._next_page_url(resp) is None
class TestOciAdapterSSRF:
def test_reject_file_scheme(self):
adapter = OciRegistryAdapter("reg.io")
with pytest.raises(ImageRegistryAuthError, match="disallowed scheme"):
adapter._validate_realm_url("file:///etc/passwd")
def test_reject_ftp_scheme(self):
adapter = OciRegistryAdapter("reg.io")
with pytest.raises(ImageRegistryAuthError, match="disallowed scheme"):
adapter._validate_realm_url("ftp://evil.com/token")
def test_reject_private_ip(self):
adapter = OciRegistryAdapter("reg.io")
with pytest.raises(ImageRegistryAuthError, match="private/loopback"):
adapter._validate_realm_url("https://10.0.0.1/token")
def test_reject_loopback(self):
adapter = OciRegistryAdapter("reg.io")
with pytest.raises(ImageRegistryAuthError, match="private/loopback"):
adapter._validate_realm_url("https://127.0.0.1/token")
def test_reject_link_local(self):
adapter = OciRegistryAdapter("reg.io")
with pytest.raises(ImageRegistryAuthError, match="private/loopback"):
adapter._validate_realm_url("https://169.254.169.254/latest/meta-data")
def test_accept_public_https(self):
adapter = OciRegistryAdapter("reg.io")
# Should not raise
adapter._validate_realm_url("https://auth.example.com/token")
def test_accept_hostname_not_ip(self):
adapter = OciRegistryAdapter("reg.io")
# Hostnames (not IPs) should pass even if they resolve to private IPs
adapter._validate_realm_url("https://internal.corp.com/token")
class TestOciAdapterEmptyToken:
@patch("prowler.providers.image.lib.registry.base.requests.request")
def test_empty_bearer_token_raises(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": "", "access_token": ""}
mock_request.side_effect = [ping_resp, token_resp]
adapter = OciRegistryAdapter("reg.io", username="u", password="p")
with pytest.raises(ImageRegistryAuthError, match="empty token"):
adapter._ensure_auth()
@patch("prowler.providers.image.lib.registry.base.requests.request")
def test_none_bearer_token_raises(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 = {}
mock_request.side_effect = [ping_resp, token_resp]
adapter = OciRegistryAdapter("reg.io", username="u", password="p")
with pytest.raises(ImageRegistryAuthError, match="empty token"):
adapter._ensure_auth()
class TestOciAdapterNarrowExcept:
def test_invalid_utf8_base64_falls_through(self):
# Create a base64 string that decodes to invalid UTF-8
invalid_bytes = base64.b64encode(b"\xff\xfe").decode()
adapter = OciRegistryAdapter("ecr.aws", username="AWS", password=invalid_bytes)
user, pwd = adapter._resolve_basic_credentials()
assert user == "AWS"
assert pwd == invalid_bytes
class TestCredentialRedaction:
def test_getstate_redacts_credentials(self):
adapter = OciRegistryAdapter("reg.io", username="u", password="secret", token="tok")
state = adapter.__getstate__()
assert state["_password"] == "***"
assert state["_token"] == "***"
assert state["username"] == "u"
assert state["registry_url"] == "reg.io"
def test_getstate_none_credentials(self):
adapter = OciRegistryAdapter("reg.io")
state = adapter.__getstate__()
assert state["_password"] is None
assert state["_token"] is None
def test_repr_redacts_credentials(self):
adapter = OciRegistryAdapter("reg.io", username="u", password="s3cret_pw", token="s3cret_tk")
r = repr(adapter)
assert "s3cret_pw" not in r
assert "s3cret_tk" not in r
assert "<redacted>" in r
def test_properties_still_work(self):
adapter = OciRegistryAdapter("reg.io", password="secret", token="tok")
assert adapter.password == "secret"
assert adapter.token == "tok"
@@ -8,6 +8,7 @@ from prowler.providers.image.exceptions.exceptions import (
ImageMaxImagesExceededError,
)
from prowler.providers.image.image_provider import ImageProvider
from prowler.providers.image.lib.registry.dockerhub_adapter import DockerHubAdapter
_CLEAN_ENV = {
"PATH": os.environ.get("PATH", ""),
@@ -215,3 +216,23 @@ class TestRegistryList:
assert exc_info.value.code == 0
captured = capsys.readouterr()
assert "6 images" in captured.out
class TestDockerHubEnumeration:
@patch("prowler.providers.image.image_provider.create_registry_adapter")
def test_dockerhub_images_use_repo_tag_format(self, mock_factory):
"""Docker Hub images should use repo:tag format without host prefix."""
adapter = MagicMock(spec=DockerHubAdapter)
adapter.list_repositories.return_value = ["myorg/app1", "myorg/app2"]
adapter.list_tags.side_effect = [["latest", "v1.0"], ["latest"]]
mock_factory.return_value = adapter
provider = _build_provider(registry="docker.io/myorg")
# Docker Hub images should NOT have host prefix
assert "myorg/app1:latest" in provider.images
assert "myorg/app1:v1.0" in provider.images
assert "myorg/app2:latest" in provider.images
# Ensure no host prefix was added
for img in provider.images:
assert not img.startswith("docker.io/"), f"Unexpected host prefix in {img}"
assert len(provider.images) == 3