ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 20:42:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(image): handle Basic auth challenge and pre-encoded ECR credentials

Browse Source 
- Detect Basic auth challenges (non-Bearer) in _ensure_auth()
- Decode base64-wrapped credentials from aws ecr get-authorization-token
- Add unit tests for credential resolution and Basic auth flow
This commit is contained in:
Andoni A.
2026-02-16 17:50:05 +01:00
parent aa284b8b24
commit 326def4a17
2 changed files with 142 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/image/lib/registry/oci_adapter.py
+36 -1
View File
@@ -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)
tests/providers/image/lib/registry/test_oci_adapter.py
+106 -7
View File
@@ -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": '<https://reg.io/v2/_catalog?n=200&last=b>; rel="next"'})
page1_resp = MagicMock(
status_code=200,
headers={"Link": '<https://reg.io/v2/_catalog?n=200&last=b>; 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": '<https://reg.io/v2/_catalog?n=200&last=b>; rel="next"'})
assert OciRegistryAdapter._next_page_url(resp) == "https://reg.io/v2/_catalog?n=200&last=b"
resp = MagicMock(
headers={"Link": '<https://reg.io/v2/_catalog?n=200&last=b>; 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": '<https://reg.io/v2/_catalog?n=200>; rel="prev"'})
resp = MagicMock(
headers={"Link": '<https://reg.io/v2/_catalog?n=200>; rel="prev"'}
)
assert OciRegistryAdapter._next_page_url(resp) is None