From ed97fcd059af7d77a9be6dbe64830dfd1162637e Mon Sep 17 00:00:00 2001 From: "Andoni A." <14891798+andoniaf@users.noreply.github.com> Date: Tue, 17 Feb 2026 10:59:49 +0100 Subject: [PATCH] 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. --- prowler/providers/image/image_provider.py | 5 +-- .../image/lib/registry/oci_adapter.py | 13 ++++++ .../image/lib/registry/test_oci_adapter.py | 42 +++++++++++++++++++ 3 files changed, 56 insertions(+), 4 deletions(-) diff --git a/prowler/providers/image/image_provider.py b/prowler/providers/image/image_provider.py index c754585f30..7a2fe2ac89 100644 --- a/prowler/providers/image/image_provider.py +++ b/prowler/providers/image/image_provider.py @@ -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 = [] diff --git a/prowler/providers/image/lib/registry/oci_adapter.py b/prowler/providers/image/lib/registry/oci_adapter.py index 9b5f97aee8..c5f1d28bcc 100644 --- a/prowler/providers/image/lib/registry/oci_adapter.py +++ b/prowler/providers/image/lib/registry/oci_adapter.py @@ -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}" diff --git a/tests/providers/image/lib/registry/test_oci_adapter.py b/tests/providers/image/lib/registry/test_oci_adapter.py index 72ca9228e4..26b13594eb 100644 --- a/tests/providers/image/lib/registry/test_oci_adapter.py +++ b/tests/providers/image/lib/registry/test_oci_adapter.py @@ -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")