mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-16 17:11:58 +00:00
fix(image): add None guard, top-level import, and 401-retry to registry adapters
Add early return in _resolve_basic_credentials when password is None to prevent TypeError on base64.b64decode. Move DockerHubAdapter import from local scope to top-level in image_provider.py per PEP 8. Add one-time 401 retry with token refresh in _authed_request to handle per-repository-scoped bearer tokens on strict OCI registries.
This commit is contained in:
@@ -41,6 +41,7 @@ from prowler.providers.image.lib.arguments.arguments import (
|
||||
SCANNERS_CHOICES,
|
||||
SEVERITY_CHOICES,
|
||||
)
|
||||
from prowler.providers.image.lib.registry.dockerhub_adapter import DockerHubAdapter
|
||||
from prowler.providers.image.lib.registry.factory import create_registry_adapter
|
||||
|
||||
|
||||
@@ -814,10 +815,6 @@ class ImageProvider(Provider):
|
||||
return
|
||||
|
||||
# Determine if this is a Docker Hub adapter (for image reference format)
|
||||
from prowler.providers.image.lib.registry.dockerhub_adapter import (
|
||||
DockerHubAdapter,
|
||||
)
|
||||
|
||||
is_dockerhub = isinstance(adapter, DockerHubAdapter)
|
||||
|
||||
discovered_images = []
|
||||
|
||||
@@ -153,6 +153,8 @@ class OciRegistryAdapter(RegistryAdapter):
|
||||
Returns (username, password) — decoded if the password is a base64 token
|
||||
containing 'username:real_password', otherwise returned as-is.
|
||||
"""
|
||||
if not self.password:
|
||||
return self.username, self.password
|
||||
try:
|
||||
decoded = base64.b64decode(self.password).decode("utf-8")
|
||||
if decoded.startswith(f"{self.username}:"):
|
||||
@@ -162,6 +164,17 @@ class OciRegistryAdapter(RegistryAdapter):
|
||||
return self.username, self.password
|
||||
|
||||
def _authed_request(self, method: str, url: str, **kwargs) -> requests.Response:
|
||||
resp = self._do_authed_request(method, url, **kwargs)
|
||||
if resp.status_code == 401 and self._bearer_token:
|
||||
logger.debug(
|
||||
f"Bearer token rejected (HTTP 401), re-authenticating to {self.registry_url}"
|
||||
)
|
||||
self._bearer_token = None
|
||||
self._ensure_auth()
|
||||
resp = self._do_authed_request(method, url, **kwargs)
|
||||
return resp
|
||||
|
||||
def _do_authed_request(self, method: str, url: str, **kwargs) -> requests.Response:
|
||||
headers = kwargs.pop("headers", {})
|
||||
if self._bearer_token:
|
||||
headers["Authorization"] = f"Bearer {self._bearer_token}"
|
||||
|
||||
@@ -153,6 +153,48 @@ class TestOciAdapterAuth:
|
||||
call_kwargs = mock_request.call_args_list[1][1]
|
||||
assert call_kwargs.get("auth") == ("AWS", raw_password)
|
||||
|
||||
def test_resolve_basic_credentials_none_password(self):
|
||||
adapter = OciRegistryAdapter("ecr.aws", username="AWS", password=None)
|
||||
user, pwd = adapter._resolve_basic_credentials()
|
||||
assert user == "AWS"
|
||||
assert pwd is None
|
||||
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_authed_request_retries_on_401_with_bearer(self, mock_request):
|
||||
adapter = OciRegistryAdapter("reg.io", username="u", password="p")
|
||||
adapter._bearer_token = "expired-token"
|
||||
# First request: 401 (expired token)
|
||||
resp_401 = MagicMock(status_code=401)
|
||||
# _ensure_auth ping: 401 with bearer challenge
|
||||
ping_resp = MagicMock(
|
||||
status_code=401,
|
||||
headers={
|
||||
"Www-Authenticate": 'Bearer realm="https://auth.reg.io/token",service="registry"'
|
||||
},
|
||||
)
|
||||
# Token exchange: success
|
||||
token_resp = MagicMock(status_code=200)
|
||||
token_resp.json.return_value = {"token": "new-token"}
|
||||
# Second request: 200 (new token works)
|
||||
resp_200 = MagicMock(status_code=200)
|
||||
mock_request.side_effect = [resp_401, ping_resp, token_resp, resp_200]
|
||||
result = adapter._authed_request("GET", "https://reg.io/v2/myapp/tags/list")
|
||||
assert result.status_code == 200
|
||||
assert adapter._bearer_token == "new-token"
|
||||
assert mock_request.call_count == 4
|
||||
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_authed_request_no_retry_on_401_without_bearer(self, mock_request):
|
||||
adapter = OciRegistryAdapter("reg.io", username="u", password="p")
|
||||
adapter._basic_auth_verified = True
|
||||
# No bearer token — using basic auth
|
||||
resp_401 = MagicMock(status_code=401)
|
||||
mock_request.return_value = resp_401
|
||||
result = adapter._authed_request("GET", "https://reg.io/v2/_catalog")
|
||||
assert result.status_code == 401
|
||||
# Should only be called once (no retry for basic auth)
|
||||
assert mock_request.call_count == 1
|
||||
|
||||
|
||||
class TestOciAdapterListRepositories:
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
|
||||
Reference in New Issue
Block a user