From b48b38162432002989a6e5048a71998e87bd402b Mon Sep 17 00:00:00 2001 From: "Andoni A." <14891798+andoniaf@users.noreply.github.com> Date: Fri, 6 Feb 2026 08:18:05 +0100 Subject: [PATCH] feat(image): add private registry authentication support Add registry_username, registry_password, and registry_token params to ImageProvider following the IaC provider pattern (explicit params with env var fallback). Credentials are injected as environment variables into Trivy subprocess calls. --- prowler/providers/image/image_provider.py | 53 +++++- .../image/lib/arguments/arguments.py | 24 +++ tests/providers/image/image_provider_test.py | 158 ++++++++++++++++++ 3 files changed, 233 insertions(+), 2 deletions(-) diff --git a/prowler/providers/image/image_provider.py b/prowler/providers/image/image_provider.py index 24f23dc02d..ded07c909a 100644 --- a/prowler/providers/image/image_provider.py +++ b/prowler/providers/image/image_provider.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import os import subprocess import sys from typing import Generator, List @@ -48,6 +49,9 @@ class ImageProvider(Provider): config_path: str = None, config_content: dict = None, fixer_config: dict | None = None, + registry_username: str = None, + registry_password: str = None, + registry_token: str = None, ): logger.info("Instantiating Image Provider...") @@ -61,7 +65,20 @@ class ImageProvider(Provider): self.audited_account = "image-scan" self._session = None self._identity = "prowler" - self._auth_method = "No auth" + + # Registry authentication (follows IaC pattern: explicit params, env vars internal) + self.registry_username = registry_username or os.environ.get("TRIVY_USERNAME") + self.registry_password = registry_password or os.environ.get("TRIVY_PASSWORD") + self.registry_token = registry_token or os.environ.get("TRIVY_REGISTRY_TOKEN") + + if self.registry_username and self.registry_password: + self._auth_method = "Basic auth" + logger.info("Using registry username/password for authentication") + elif self.registry_token: + self._auth_method = "Registry token" + logger.info("Using registry token for authentication") + else: + self._auth_method = "No auth" # Load images from file if provided if image_list_file: @@ -394,8 +411,19 @@ class ImageProvider(Provider): ) logger.error(f"Error scanning image {image}: {error}") + def _build_trivy_env(self) -> dict: + """Build environment variables for Trivy, injecting registry credentials.""" + env = dict(os.environ) + if self.registry_username and self.registry_password: + env["TRIVY_USERNAME"] = self.registry_username + env["TRIVY_PASSWORD"] = self.registry_password + elif self.registry_token: + env["TRIVY_REGISTRY_TOKEN"] = self.registry_token + return env + def _execute_trivy(self, command: list, image: str) -> subprocess.CompletedProcess: """Execute Trivy command with optional progress bar.""" + env = self._build_trivy_env() try: if sys.stdout.isatty(): with alive_bar( @@ -410,6 +438,7 @@ class ImageProvider(Provider): command, capture_output=True, text=True, + env=env, ) bar.title = f"-> Scan completed for {image}" return process @@ -419,12 +448,13 @@ class ImageProvider(Provider): command, capture_output=True, text=True, + env=env, ) logger.info(f"Scan completed for {image}") return process except (AttributeError, OSError): logger.info(f"Scanning {image}...") - return subprocess.run(command, capture_output=True, text=True) + return subprocess.run(command, capture_output=True, text=True, env=env) def _log_trivy_stderr(self, stderr: str) -> None: """Parse and log Trivy's stderr output.""" @@ -474,6 +504,10 @@ class ImageProvider(Provider): report_lines.append(f"Timeout: {Fore.YELLOW}{self.timeout}{Style.RESET_ALL}") + report_lines.append( + f"Authentication method: {Fore.YELLOW}{self.auth_method}{Style.RESET_ALL}" + ) + print_boxes(report_lines, report_title) @staticmethod @@ -481,6 +515,9 @@ class ImageProvider(Provider): image: str = None, raise_on_exception: bool = True, provider_id: str = None, + registry_username: str = None, + registry_password: str = None, + registry_token: str = None, ) -> "Connection": """ Test connection to container registry by attempting to inspect an image. @@ -489,6 +526,9 @@ class ImageProvider(Provider): image: Container image to test raise_on_exception: Whether to raise exceptions provider_id: Fallback for image name + registry_username: Registry username for basic auth + registry_password: Registry password for basic auth + registry_token: Registry token for token-based auth Returns: Connection: Connection object with success status @@ -500,6 +540,14 @@ class ImageProvider(Provider): if not image: return Connection(is_connected=False, error="Image name is required") + # Build env with registry credentials + env = dict(os.environ) + if registry_username and registry_password: + env["TRIVY_USERNAME"] = registry_username + env["TRIVY_PASSWORD"] = registry_password + elif registry_token: + env["TRIVY_REGISTRY_TOKEN"] = registry_token + # Test by running trivy with --skip-update to just test image access process = subprocess.run( [ @@ -512,6 +560,7 @@ class ImageProvider(Provider): capture_output=True, text=True, timeout=60, + env=env, ) if process.returncode == 0: diff --git a/prowler/providers/image/lib/arguments/arguments.py b/prowler/providers/image/lib/arguments/arguments.py index e27efac97e..eec5414190 100644 --- a/prowler/providers/image/lib/arguments/arguments.py +++ b/prowler/providers/image/lib/arguments/arguments.py @@ -74,6 +74,30 @@ def init_parser(self): help="Trivy scan timeout. Default: 5m. Examples: 10m, 1h", ) + # Registry Authentication + registry_auth_group = image_parser.add_argument_group("Registry Authentication") + registry_auth_group.add_argument( + "--registry-username", + dest="registry_username", + nargs="?", + default=None, + help="Username for private registry authentication (used with --registry-password). If not provided, will use TRIVY_USERNAME env var.", + ) + registry_auth_group.add_argument( + "--registry-password", + dest="registry_password", + nargs="?", + default=None, + help="Password for private registry authentication (used with --registry-username). If not provided, will use TRIVY_PASSWORD env var.", + ) + registry_auth_group.add_argument( + "--registry-token", + dest="registry_token", + nargs="?", + default=None, + help="Token for private registry authentication. If not provided, will use TRIVY_REGISTRY_TOKEN env var.", + ) + def validate_arguments(arguments): """Validate Image provider arguments.""" diff --git a/tests/providers/image/image_provider_test.py b/tests/providers/image/image_provider_test.py index 1ee91f5c90..9b4e06eac9 100644 --- a/tests/providers/image/image_provider_test.py +++ b/tests/providers/image/image_provider_test.py @@ -1,3 +1,4 @@ +import os import tempfile from unittest import mock from unittest.mock import MagicMock, patch @@ -352,3 +353,160 @@ class TestImageProvider: assert isinstance(reports, list) assert len(reports) == 1 + + +class TestImageProviderRegistryAuth: + def test_no_auth_by_default(self): + """Test that no auth is set when no credentials are provided.""" + provider = _make_provider() + + assert provider.registry_username is None + assert provider.registry_password is None + assert provider.registry_token is None + assert provider.auth_method == "No auth" + + def test_basic_auth_with_explicit_params(self): + """Test basic auth via explicit constructor params.""" + provider = _make_provider( + registry_username="myuser", + registry_password="mypass", + ) + + assert provider.registry_username == "myuser" + assert provider.registry_password == "mypass" + assert provider.auth_method == "Basic auth" + + def test_token_auth_with_explicit_param(self): + """Test token auth via explicit constructor param.""" + provider = _make_provider(registry_token="my-token-123") + + assert provider.registry_token == "my-token-123" + assert provider.auth_method == "Registry token" + + def test_basic_auth_takes_precedence_over_token(self): + """Test that username/password takes precedence over token.""" + provider = _make_provider( + registry_username="myuser", + registry_password="mypass", + registry_token="my-token", + ) + + assert provider.auth_method == "Basic auth" + + @patch.dict(os.environ, {"TRIVY_USERNAME": "envuser", "TRIVY_PASSWORD": "envpass"}) + def test_basic_auth_from_env_vars(self): + """Test that env vars are used as fallback for basic auth.""" + provider = _make_provider() + + assert provider.registry_username == "envuser" + assert provider.registry_password == "envpass" + assert provider.auth_method == "Basic auth" + + @patch.dict(os.environ, {"TRIVY_REGISTRY_TOKEN": "env-token"}) + def test_token_auth_from_env_var(self): + """Test that env var is used as fallback for token auth.""" + provider = _make_provider() + + assert provider.registry_token == "env-token" + assert provider.auth_method == "Registry token" + + @patch.dict(os.environ, {"TRIVY_USERNAME": "envuser", "TRIVY_PASSWORD": "envpass"}) + def test_explicit_params_override_env_vars(self): + """Test that explicit params take precedence over env vars.""" + provider = _make_provider( + registry_username="explicit", + registry_password="explicit-pass", + ) + + assert provider.registry_username == "explicit" + assert provider.registry_password == "explicit-pass" + + def test_build_trivy_env_no_auth(self): + """Test that _build_trivy_env returns base env when no auth.""" + provider = _make_provider() + env = provider._build_trivy_env() + + assert "TRIVY_USERNAME" not in env + assert "TRIVY_PASSWORD" not in env + assert "TRIVY_REGISTRY_TOKEN" not in env + + def test_build_trivy_env_basic_auth(self): + """Test that _build_trivy_env injects username/password.""" + provider = _make_provider( + registry_username="myuser", + registry_password="mypass", + ) + env = provider._build_trivy_env() + + assert env["TRIVY_USERNAME"] == "myuser" + assert env["TRIVY_PASSWORD"] == "mypass" + + def test_build_trivy_env_token_auth(self): + """Test that _build_trivy_env injects registry token.""" + provider = _make_provider(registry_token="my-token") + env = provider._build_trivy_env() + + assert env["TRIVY_REGISTRY_TOKEN"] == "my-token" + + @patch("subprocess.run") + def test_execute_trivy_passes_env(self, mock_subprocess): + """Test that _execute_trivy passes credentials via env.""" + provider = _make_provider( + registry_username="myuser", + registry_password="mypass", + ) + mock_subprocess.return_value = MagicMock( + stdout=get_sample_trivy_json_output(), stderr="" + ) + + provider._execute_trivy(["trivy", "image", "alpine:3.18"], "alpine:3.18") + + call_kwargs = mock_subprocess.call_args + env = call_kwargs.kwargs.get("env") or call_kwargs[1].get("env") + assert env["TRIVY_USERNAME"] == "myuser" + assert env["TRIVY_PASSWORD"] == "mypass" + + @patch("subprocess.run") + def test_test_connection_with_basic_auth(self, mock_subprocess): + """Test test_connection passes credentials via env.""" + mock_subprocess.return_value = MagicMock(returncode=0, stderr="") + + result = ImageProvider.test_connection( + image="private.registry.io/myapp:v1", + registry_username="myuser", + registry_password="mypass", + ) + + assert result.is_connected is True + call_kwargs = mock_subprocess.call_args + env = call_kwargs.kwargs.get("env") or call_kwargs[1].get("env") + assert env["TRIVY_USERNAME"] == "myuser" + assert env["TRIVY_PASSWORD"] == "mypass" + + @patch("subprocess.run") + def test_test_connection_with_token(self, mock_subprocess): + """Test test_connection passes token via env.""" + mock_subprocess.return_value = MagicMock(returncode=0, stderr="") + + result = ImageProvider.test_connection( + image="private.registry.io/myapp:v1", + registry_token="my-token", + ) + + assert result.is_connected is True + call_kwargs = mock_subprocess.call_args + env = call_kwargs.kwargs.get("env") or call_kwargs[1].get("env") + assert env["TRIVY_REGISTRY_TOKEN"] == "my-token" + + def test_print_credentials_shows_auth_method(self): + """Test that print_credentials outputs the auth method.""" + provider = _make_provider( + registry_username="myuser", + registry_password="mypass", + ) + with mock.patch("builtins.print") as mock_print: + provider.print_credentials() + output = " ".join( + str(call.args[0]) for call in mock_print.call_args_list if call.args + ) + assert "Basic auth" in output