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.
This commit is contained in:
Andoni A.
2026-02-06 08:18:05 +01:00
parent 9d5e3f4758
commit b48b381624
3 changed files with 233 additions and 2 deletions
+51 -2
View File
@@ -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:
@@ -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."""
@@ -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