fix(image): harden registry adapter layer

- Fix shared dict mutation in ImageBaseException (thread-safety)
- Handle unexpected status codes in OCI adapter auth flow
- Force TLS verification for Docker Hub endpoints
- Add credential-safe __repr__ to RegistryAdapter
- Rename _next_page_url override to _next_tag_page_url in DockerHubAdapter
- Distinguish rate-limit from connection errors in retry logic
- Add type hints to all public adapter methods
- Remove unused imports in test files (flake8)
This commit is contained in:
Andoni A. committed 2026-02-17 10:01:35 +01:00
1 parent b00c76acf2
commit d9e451489d
8 files changed
+251 -73

No files matched your search

@@ -1,90 +1,210 @@
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)
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)
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)
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)
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)
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)
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)
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)
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)
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)
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)
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)
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_registry_list_without_registry_fails(self):
args = Namespace(images=["nginx:latest"], image_list_file=None, registry=None, image_filter=None, tag_filter=None, max_images=0, registry_insecure=False, registry_list_images=True)
args = Namespace(
images=["nginx:latest"],
image_list_file=None,
registry=None,
image_filter=None,
tag_filter=None,
max_images=0,
registry_insecure=False,
registry_list_images=True,
)
ok, msg = validate_arguments(args)
assert not ok
assert "--registry-list requires --registry" in msg
def test_registry_list_with_registry_passes(self):
args = Namespace(images=[], image_list_file=None, registry="myregistry.io", image_filter=None, tag_filter=None, max_images=0, registry_insecure=False, registry_list_images=True)
args = Namespace(
images=[],
image_list_file=None,
registry="myregistry.io",
image_filter=None,
tag_filter=None,
max_images=0,
registry_insecure=False,
registry_list_images=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)
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
@@ -1,7 +1,3 @@
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,
@@ -27,10 +23,15 @@ class TestDetectRegistryType:
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"
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"
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"
@@ -51,7 +52,11 @@ class TestCreateRegistryAdapter:
def test_passes_credentials(self):
adapter = create_registry_adapter(
"myregistry.io", username="user", password="pass", token="tok", verify_ssl=False
"myregistry.io",
username="user",
password="pass",
token="tok",
verify_ssl=False,
)
assert adapter.username == "user"
assert adapter.password == "pass"
@@ -1,5 +1,4 @@
import os
from argparse import Namespace
from unittest.mock import MagicMock, patch
import pytest