diff --git a/prowler/providers/image/lib/registry/oci_adapter.py b/prowler/providers/image/lib/registry/oci_adapter.py index 3fbeb64018..310a84cd37 100644 --- a/prowler/providers/image/lib/registry/oci_adapter.py +++ b/prowler/providers/image/lib/registry/oci_adapter.py @@ -2,6 +2,7 @@ from __future__ import annotations +import base64 import re import time @@ -28,6 +29,7 @@ class OciRegistryAdapter(RegistryAdapter): super().__init__(registry_url, username, password, token, verify_ssl) self._base_url = self._normalise_url(registry_url) self._bearer_token = None + self._basic_auth_verified = False @staticmethod def _normalise_url(url): @@ -72,6 +74,8 @@ class OciRegistryAdapter(RegistryAdapter): def _ensure_auth(self, repository=None): if self._bearer_token: return + if self._basic_auth_verified: + return if self.token: self._bearer_token = self.token return @@ -81,6 +85,22 @@ class OciRegistryAdapter(RegistryAdapter): return if resp.status_code == 401: www_auth = resp.headers.get("Www-Authenticate", "") + + if not www_auth.lower().startswith("bearer"): + # Basic auth challenge (e.g., AWS ECR) + if self.username and self.password: + self._basic_auth_verified = True + return + raise ImageRegistryAuthError( + file=__file__, + message=( + f"Registry {self.registry_url} requires authentication " + f"but no credentials provided. " + f"Set REGISTRY_USERNAME and REGISTRY_PASSWORD." + ), + ) + + # Bearer token exchange (standard OCI flow) self._bearer_token = self._obtain_bearer_token(www_auth, repository) return if resp.status_code == 403: @@ -118,12 +138,27 @@ class OciRegistryAdapter(RegistryAdapter): data = resp.json() return data.get("token") or data.get("access_token", "") + def _resolve_basic_credentials(self): + """Decode pre-encoded base64 auth tokens (e.g., from aws ecr get-authorization-token). + + Returns (username, password) — decoded if the password is a base64 token + containing 'username:real_password', otherwise returned as-is. + """ + try: + decoded = base64.b64decode(self.password).decode("utf-8") + if decoded.startswith(f"{self.username}:"): + return self.username, decoded[len(self.username) + 1 :] + except Exception: + pass + return self.username, self.password + def _authed_request(self, method, url, **kwargs): headers = kwargs.pop("headers", {}) if self._bearer_token: headers["Authorization"] = f"Bearer {self._bearer_token}" elif self.username and self.password: - kwargs.setdefault("auth", (self.username, self.password)) + user, pwd = self._resolve_basic_credentials() + kwargs.setdefault("auth", (user, pwd)) kwargs["headers"] = headers return self._request_with_retry(method, url, **kwargs) diff --git a/tests/providers/image/lib/registry/test_oci_adapter.py b/tests/providers/image/lib/registry/test_oci_adapter.py index 385da5cea9..9285ffe198 100644 --- a/tests/providers/image/lib/registry/test_oci_adapter.py +++ b/tests/providers/image/lib/registry/test_oci_adapter.py @@ -1,3 +1,4 @@ +import base64 from unittest.mock import MagicMock, patch import pytest @@ -25,7 +26,9 @@ class TestOciAdapterInit: 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) + 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" @@ -50,7 +53,12 @@ class TestOciAdapterAuth: @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"'}) + 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] @@ -66,13 +74,94 @@ class TestOciAdapterAuth: with pytest.raises(ImageRegistryAuthError): adapter._ensure_auth() + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_ensure_auth_basic_challenge_with_creds(self, mock_request): + ping_resp = MagicMock( + status_code=401, + headers={"Www-Authenticate": 'Basic realm="https://ecr.aws"'}, + ) + mock_request.return_value = ping_resp + adapter = OciRegistryAdapter("ecr.aws", username="AWS", password="tok") + adapter._ensure_auth() + assert adapter._basic_auth_verified is True + assert adapter._bearer_token is None + + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_ensure_auth_basic_challenge_no_creds(self, mock_request): + ping_resp = MagicMock( + status_code=401, + headers={"Www-Authenticate": 'Basic realm="https://ecr.aws"'}, + ) + mock_request.return_value = ping_resp + adapter = OciRegistryAdapter("ecr.aws") + with pytest.raises(ImageRegistryAuthError): + adapter._ensure_auth() + + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_basic_auth_used_in_requests(self, mock_request): + ping_resp = MagicMock( + status_code=401, + headers={"Www-Authenticate": 'Basic realm="https://ecr.aws"'}, + ) + catalog_resp = MagicMock(status_code=200, headers={}) + catalog_resp.json.return_value = {"repositories": ["myapp"]} + mock_request.side_effect = [ping_resp, catalog_resp] + adapter = OciRegistryAdapter("ecr.aws", username="AWS", password="tok") + adapter._ensure_auth() + adapter._authed_request("GET", "https://ecr.aws/v2/_catalog") + # The catalog request should use Basic auth (auth kwarg), not Bearer header + call_kwargs = mock_request.call_args_list[1][1] + assert call_kwargs.get("auth") == ("AWS", "tok") + assert "Authorization" not in call_kwargs.get("headers", {}) + + def test_resolve_basic_credentials_decodes_base64_token(self): + raw_password = "real-jwt-password" + encoded = base64.b64encode(f"AWS:{raw_password}".encode()).decode() + adapter = OciRegistryAdapter("ecr.aws", username="AWS", password=encoded) + user, pwd = adapter._resolve_basic_credentials() + assert user == "AWS" + assert pwd == raw_password + + def test_resolve_basic_credentials_passthrough_raw_password(self): + adapter = OciRegistryAdapter("ecr.aws", username="AWS", password="plain-pass") + user, pwd = adapter._resolve_basic_credentials() + assert user == "AWS" + assert pwd == "plain-pass" + + def test_resolve_basic_credentials_passthrough_invalid_base64(self): + adapter = OciRegistryAdapter( + "ecr.aws", username="AWS", password="not!valid~base64" + ) + user, pwd = adapter._resolve_basic_credentials() + assert user == "AWS" + assert pwd == "not!valid~base64" + + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_basic_auth_decodes_ecr_token_in_request(self, mock_request): + raw_password = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0In0.abc" + encoded = base64.b64encode(f"AWS:{raw_password}".encode()).decode() + ping_resp = MagicMock( + status_code=401, + headers={"Www-Authenticate": 'Basic realm="https://ecr.aws"'}, + ) + catalog_resp = MagicMock(status_code=200, headers={}) + catalog_resp.json.return_value = {"repositories": ["myapp"]} + mock_request.side_effect = [ping_resp, catalog_resp] + adapter = OciRegistryAdapter("ecr.aws", username="AWS", password=encoded) + adapter._ensure_auth() + adapter._authed_request("GET", "https://ecr.aws/v2/_catalog") + call_kwargs = mock_request.call_args_list[1][1] + assert call_kwargs.get("auth") == ("AWS", raw_password) + 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"]} + 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() @@ -81,7 +170,10 @@ class TestOciAdapterListRepositories: @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": '; rel="next"'}) + page1_resp = MagicMock( + status_code=200, + headers={"Link": '; rel="next"'}, + ) page1_resp.json.return_value = {"repositories": ["a"]} page2_resp = MagicMock(status_code=200, headers={}) page2_resp.json.return_value = {"repositories": ["b"]} @@ -158,9 +250,16 @@ class TestOciAdapterNextPageUrl: assert OciRegistryAdapter._next_page_url(resp) is None def test_link_header_with_next(self): - resp = MagicMock(headers={"Link": '; rel="next"'}) - assert OciRegistryAdapter._next_page_url(resp) == "https://reg.io/v2/_catalog?n=200&last=b" + resp = MagicMock( + headers={"Link": '; 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": '; rel="prev"'}) + resp = MagicMock( + headers={"Link": '; rel="prev"'} + ) assert OciRegistryAdapter._next_page_url(resp) is None