diff --git a/api/Dockerfile b/api/Dockerfile index 2d7883a957..b6d5d04fcf 100644 --- a/api/Dockerfile +++ b/api/Dockerfile @@ -5,7 +5,7 @@ LABEL maintainer="https://github.com/prowler-cloud/api" ARG POWERSHELL_VERSION=7.5.0 ENV POWERSHELL_VERSION=${POWERSHELL_VERSION} -ARG TRIVY_VERSION=0.66.0 +ARG TRIVY_VERSION=0.69.1 ENV TRIVY_VERSION=${TRIVY_VERSION} # hadolint ignore=DL3008 diff --git a/api/src/backend/api/migrations/0076_image_provider.py b/api/src/backend/api/migrations/0076_image_provider.py new file mode 100644 index 0000000000..9f197bf610 --- /dev/null +++ b/api/src/backend/api/migrations/0076_image_provider.py @@ -0,0 +1,37 @@ +from django.db import migrations + +import api.db_utils + + +class Migration(migrations.Migration): + dependencies = [ + ("api", "0075_cloudflare_provider"), + ] + + operations = [ + migrations.AlterField( + model_name="provider", + name="provider", + field=api.db_utils.ProviderEnumField( + choices=[ + ("aws", "AWS"), + ("azure", "Azure"), + ("gcp", "GCP"), + ("kubernetes", "Kubernetes"), + ("m365", "M365"), + ("github", "GitHub"), + ("mongodbatlas", "MongoDB Atlas"), + ("iac", "IaC"), + ("oraclecloud", "Oracle Cloud Infrastructure"), + ("alibabacloud", "Alibaba Cloud"), + ("cloudflare", "Cloudflare"), + ("image", "Image"), + ], + default="aws", + ), + ), + migrations.RunSQL( + "ALTER TYPE provider ADD VALUE IF NOT EXISTS 'image';", + reverse_sql=migrations.RunSQL.noop, + ), + ] diff --git a/poetry.lock b/poetry.lock index b3fb2a314c..f4d82258e7 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.3.1 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.1.1 and should not be changed by hand. [[package]] name = "about-time" @@ -5874,6 +5874,7 @@ files = [ {file = "ruamel.yaml.clib-0.2.12-cp310-cp310-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:f66efbc1caa63c088dead1c4170d148eabc9b80d95fb75b6c92ac0aad2437d76"}, {file = "ruamel.yaml.clib-0.2.12-cp310-cp310-musllinux_1_1_i686.whl", hash = "sha256:22353049ba4181685023b25b5b51a574bce33e7f51c759371a7422dcae5402a6"}, {file = "ruamel.yaml.clib-0.2.12-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:932205970b9f9991b34f55136be327501903f7c66830e9760a8ffb15b07f05cd"}, + {file = "ruamel.yaml.clib-0.2.12-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:a52d48f4e7bf9005e8f0a89209bf9a73f7190ddf0489eee5eb51377385f59f2a"}, {file = "ruamel.yaml.clib-0.2.12-cp310-cp310-win32.whl", hash = "sha256:3eac5a91891ceb88138c113f9db04f3cebdae277f5d44eaa3651a4f573e6a5da"}, {file = "ruamel.yaml.clib-0.2.12-cp310-cp310-win_amd64.whl", hash = "sha256:ab007f2f5a87bd08ab1499bdf96f3d5c6ad4dcfa364884cb4549aa0154b13a28"}, {file = "ruamel.yaml.clib-0.2.12-cp311-cp311-macosx_13_0_arm64.whl", hash = "sha256:4a6679521a58256a90b0d89e03992c15144c5f3858f40d7c18886023d7943db6"}, @@ -5882,6 +5883,7 @@ files = [ {file = "ruamel.yaml.clib-0.2.12-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:811ea1594b8a0fb466172c384267a4e5e367298af6b228931f273b111f17ef52"}, {file = "ruamel.yaml.clib-0.2.12-cp311-cp311-musllinux_1_1_i686.whl", hash = "sha256:cf12567a7b565cbf65d438dec6cfbe2917d3c1bdddfce84a9930b7d35ea59642"}, {file = "ruamel.yaml.clib-0.2.12-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:7dd5adc8b930b12c8fc5b99e2d535a09889941aa0d0bd06f4749e9a9397c71d2"}, + {file = "ruamel.yaml.clib-0.2.12-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:1492a6051dab8d912fc2adeef0e8c72216b24d57bd896ea607cb90bb0c4981d3"}, {file = "ruamel.yaml.clib-0.2.12-cp311-cp311-win32.whl", hash = "sha256:bd0a08f0bab19093c54e18a14a10b4322e1eacc5217056f3c063bd2f59853ce4"}, {file = "ruamel.yaml.clib-0.2.12-cp311-cp311-win_amd64.whl", hash = "sha256:a274fb2cb086c7a3dea4322ec27f4cb5cc4b6298adb583ab0e211a4682f241eb"}, {file = "ruamel.yaml.clib-0.2.12-cp312-cp312-macosx_14_0_arm64.whl", hash = "sha256:20b0f8dc160ba83b6dcc0e256846e1a02d044e13f7ea74a3d1d56ede4e48c632"}, @@ -5890,6 +5892,7 @@ files = [ {file = "ruamel.yaml.clib-0.2.12-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:749c16fcc4a2b09f28843cda5a193e0283e47454b63ec4b81eaa2242f50e4ccd"}, {file = "ruamel.yaml.clib-0.2.12-cp312-cp312-musllinux_1_1_i686.whl", hash = "sha256:bf165fef1f223beae7333275156ab2022cffe255dcc51c27f066b4370da81e31"}, {file = "ruamel.yaml.clib-0.2.12-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:32621c177bbf782ca5a18ba4d7af0f1082a3f6e517ac2a18b3974d4edf349680"}, + {file = "ruamel.yaml.clib-0.2.12-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:b82a7c94a498853aa0b272fd5bc67f29008da798d4f93a2f9f289feb8426a58d"}, {file = "ruamel.yaml.clib-0.2.12-cp312-cp312-win32.whl", hash = "sha256:e8c4ebfcfd57177b572e2040777b8abc537cdef58a2120e830124946aa9b42c5"}, {file = "ruamel.yaml.clib-0.2.12-cp312-cp312-win_amd64.whl", hash = "sha256:0467c5965282c62203273b838ae77c0d29d7638c8a4e3a1c8bdd3602c10904e4"}, {file = "ruamel.yaml.clib-0.2.12-cp313-cp313-macosx_14_0_arm64.whl", hash = "sha256:4c8c5d82f50bb53986a5e02d1b3092b03622c02c2eb78e29bec33fd9593bae1a"}, @@ -5898,6 +5901,7 @@ files = [ {file = "ruamel.yaml.clib-0.2.12-cp313-cp313-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:96777d473c05ee3e5e3c3e999f5d23c6f4ec5b0c38c098b3a5229085f74236c6"}, {file = "ruamel.yaml.clib-0.2.12-cp313-cp313-musllinux_1_1_i686.whl", hash = "sha256:3bc2a80e6420ca8b7d3590791e2dfc709c88ab9152c00eeb511c9875ce5778bf"}, {file = "ruamel.yaml.clib-0.2.12-cp313-cp313-musllinux_1_1_x86_64.whl", hash = "sha256:e188d2699864c11c36cdfdada94d781fd5d6b0071cd9c427bceb08ad3d7c70e1"}, + {file = "ruamel.yaml.clib-0.2.12-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:4f6f3eac23941b32afccc23081e1f50612bdbe4e982012ef4f5797986828cd01"}, {file = "ruamel.yaml.clib-0.2.12-cp313-cp313-win32.whl", hash = "sha256:6442cb36270b3afb1b4951f060eccca1ce49f3d087ca1ca4563a6eb479cb3de6"}, {file = "ruamel.yaml.clib-0.2.12-cp313-cp313-win_amd64.whl", hash = "sha256:e5b8daf27af0b90da7bb903a876477a9e6d7270be6146906b276605997c7e9a3"}, {file = "ruamel.yaml.clib-0.2.12-cp39-cp39-macosx_12_0_arm64.whl", hash = "sha256:fc4b630cd3fa2cf7fce38afa91d7cfe844a9f75d7f0f36393fa98815e911d987"}, @@ -5906,6 +5910,7 @@ files = [ {file = "ruamel.yaml.clib-0.2.12-cp39-cp39-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:e2f1c3765db32be59d18ab3953f43ab62a761327aafc1594a2a1fbe038b8b8a7"}, {file = "ruamel.yaml.clib-0.2.12-cp39-cp39-musllinux_1_1_i686.whl", hash = "sha256:d85252669dc32f98ebcd5d36768f5d4faeaeaa2d655ac0473be490ecdae3c285"}, {file = "ruamel.yaml.clib-0.2.12-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:e143ada795c341b56de9418c58d028989093ee611aa27ffb9b7f609c00d813ed"}, + {file = "ruamel.yaml.clib-0.2.12-cp39-cp39-musllinux_1_2_aarch64.whl", hash = "sha256:2c59aa6170b990d8d2719323e628aaf36f3bfbc1c26279c0eeeb24d05d2d11c7"}, {file = "ruamel.yaml.clib-0.2.12-cp39-cp39-win32.whl", hash = "sha256:beffaed67936fbbeffd10966a4eb53c402fafd3d6833770516bf7314bc6ffa12"}, {file = "ruamel.yaml.clib-0.2.12-cp39-cp39-win_amd64.whl", hash = "sha256:040ae85536960525ea62868b642bdb0c2cc6021c9f9d507810c0c604e66f5a7b"}, {file = "ruamel.yaml.clib-0.2.12.tar.gz", hash = "sha256:6c8fbb13ec503f99a91901ab46e0b07ae7941cd527393187039aec586fdfd36f"}, diff --git a/prowler/providers/image/__init__.py b/prowler/providers/image/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/image/exceptions/__init__.py b/prowler/providers/image/exceptions/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/image/exceptions/exceptions.py b/prowler/providers/image/exceptions/exceptions.py new file mode 100644 index 0000000000..387b443ce3 --- /dev/null +++ b/prowler/providers/image/exceptions/exceptions.py @@ -0,0 +1,229 @@ +from prowler.exceptions.exceptions import ProwlerException + + +# Exceptions codes from 11000 to 11999 are reserved for Image exceptions +class ImageBaseException(ProwlerException): + """Base class for Image provider errors.""" + + IMAGE_ERROR_CODES = { + (11000, "ImageNoImagesProvidedError"): { + "message": "No container images provided for scanning.", + "remediation": "Provide at least one image using --image or --image-list-file.", + }, + (11001, "ImageListFileNotFoundError"): { + "message": "Image list file not found.", + "remediation": "Ensure the image list file exists at the specified path.", + }, + (11002, "ImageListFileReadError"): { + "message": "Error reading image list file.", + "remediation": "Check file permissions and format. The file should contain one image per line.", + }, + (11003, "ImageFindingProcessingError"): { + "message": "Error processing image scan finding.", + "remediation": "Check the Trivy output format and ensure the finding structure is valid.", + }, + (11004, "ImageTrivyBinaryNotFoundError"): { + "message": "Trivy binary not found.", + "remediation": "Install Trivy from https://trivy.dev/latest/getting-started/installation/", + }, + (11005, "ImageScanError"): { + "message": "Error scanning container image.", + "remediation": "Check the image name and ensure it is accessible.", + }, + (11006, "ImageInvalidTimeoutError"): { + "message": "Invalid timeout format.", + "remediation": "Use a valid timeout like '5m', '300s', or '1h'.", + }, + (11007, "ImageInvalidScannerError"): { + "message": "Invalid scanner type.", + "remediation": "Use valid scanners: vuln, secret, misconfig, license.", + }, + (11008, "ImageInvalidSeverityError"): { + "message": "Invalid severity level.", + "remediation": "Use valid severities: CRITICAL, HIGH, MEDIUM, LOW, UNKNOWN.", + }, + (11009, "ImageInvalidNameError"): { + "message": "Invalid container image name.", + "remediation": "Use a valid image reference (e.g., 'alpine:3.18', 'registry.example.com/repo/image:tag').", + }, + (11010, "ImageInvalidConfigScannerError"): { + "message": "Invalid image config scanner type.", + "remediation": "Use valid image config scanners: misconfig, secret.", + }, + (11013, "ImageRegistryAuthError"): { + "message": "Registry authentication failed.", + "remediation": "Check REGISTRY_USERNAME/REGISTRY_PASSWORD or REGISTRY_TOKEN environment variables.", + }, + (11014, "ImageRegistryCatalogError"): { + "message": "Registry does not support catalog listing.", + "remediation": "Use --image or --image-list instead of --registry.", + }, + (11015, "ImageRegistryNetworkError"): { + "message": "Network error communicating with registry.", + "remediation": "Check registry URL and network connectivity.", + }, + (11016, "ImageMaxImagesExceededError"): { + "message": "Discovered images exceed --max-images limit.", + "remediation": "Use --image-filter or --tag-filter to narrow results, or increase --max-images.", + }, + (11017, "ImageInvalidFilterError"): { + "message": "Invalid regex filter pattern.", + "remediation": "Check the regex syntax for --image-filter or --tag-filter.", + }, + } + + def __init__(self, code, file=None, original_exception=None, message=None): + error_info = self.IMAGE_ERROR_CODES.get((code, self.__class__.__name__)) + if error_info and message: + error_info = {**error_info, "message": message} + super().__init__( + code, + source="Image", + file=file, + original_exception=original_exception, + error_info=error_info, + ) + + +class ImageNoImagesProvidedError(ImageBaseException): + """Exception raised when no container images are provided for scanning.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11000, file=file, original_exception=original_exception, message=message + ) + + +class ImageListFileNotFoundError(ImageBaseException): + """Exception raised when the image list file is not found.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11001, file=file, original_exception=original_exception, message=message + ) + + +class ImageListFileReadError(ImageBaseException): + """Exception raised when the image list file cannot be read.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11002, file=file, original_exception=original_exception, message=message + ) + + +class ImageFindingProcessingError(ImageBaseException): + """Exception raised when a finding cannot be processed.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11003, file=file, original_exception=original_exception, message=message + ) + + +class ImageTrivyBinaryNotFoundError(ImageBaseException): + """Exception raised when the Trivy binary is not found.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11004, file=file, original_exception=original_exception, message=message + ) + + +class ImageScanError(ImageBaseException): + """Exception raised when a general scan error occurs.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11005, file=file, original_exception=original_exception, message=message + ) + + +class ImageInvalidTimeoutError(ImageBaseException): + """Exception raised when an invalid timeout format is provided.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11006, file=file, original_exception=original_exception, message=message + ) + + +class ImageInvalidScannerError(ImageBaseException): + """Exception raised when an invalid scanner type is provided.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11007, file=file, original_exception=original_exception, message=message + ) + + +class ImageInvalidSeverityError(ImageBaseException): + """Exception raised when an invalid severity level is provided.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11008, file=file, original_exception=original_exception, message=message + ) + + +class ImageInvalidNameError(ImageBaseException): + """Exception raised when an invalid container image name is provided.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11009, file=file, original_exception=original_exception, message=message + ) + + +class ImageInvalidConfigScannerError(ImageBaseException): + """Exception raised when an invalid image config scanner type is provided.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11010, file=file, original_exception=original_exception, message=message + ) + + +class ImageRegistryAuthError(ImageBaseException): + """Exception raised when registry authentication fails.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11013, file=file, original_exception=original_exception, message=message + ) + + +class ImageRegistryCatalogError(ImageBaseException): + """Exception raised when registry does not support catalog listing.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11014, file=file, original_exception=original_exception, message=message + ) + + +class ImageRegistryNetworkError(ImageBaseException): + """Exception raised when a network error occurs communicating with a registry.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11015, file=file, original_exception=original_exception, message=message + ) + + +class ImageMaxImagesExceededError(ImageBaseException): + """Exception raised when discovered images exceed --max-images limit.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11016, file=file, original_exception=original_exception, message=message + ) + + +class ImageInvalidFilterError(ImageBaseException): + """Exception raised when an invalid regex filter pattern is provided.""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 11017, file=file, original_exception=original_exception, message=message + ) diff --git a/prowler/providers/image/image_provider.py b/prowler/providers/image/image_provider.py new file mode 100644 index 0000000000..be16580ba1 --- /dev/null +++ b/prowler/providers/image/image_provider.py @@ -0,0 +1,983 @@ +from __future__ import annotations + +import json +import os +import re +import subprocess +import sys +from typing import Generator + +from alive_progress import alive_bar +from colorama import Fore, Style + +from prowler.config.config import ( + default_config_file_path, + load_and_validate_config_file, +) +from prowler.lib.check.models import CheckReportImage +from prowler.lib.logger import logger +from prowler.lib.utils.utils import print_boxes +from prowler.providers.common.models import Audit_Metadata, Connection +from prowler.providers.common.provider import Provider +from prowler.providers.image.exceptions.exceptions import ( + ImageFindingProcessingError, + ImageInvalidConfigScannerError, + ImageInvalidFilterError, + ImageInvalidNameError, + ImageInvalidScannerError, + ImageInvalidSeverityError, + ImageInvalidTimeoutError, + ImageListFileNotFoundError, + ImageListFileReadError, + ImageMaxImagesExceededError, + ImageNoImagesProvidedError, + ImageScanError, + ImageTrivyBinaryNotFoundError, +) +from prowler.providers.image.lib.arguments.arguments import ( + IMAGE_CONFIG_SCANNERS_CHOICES, + 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 + + +class ImageProvider(Provider): + """ + Container Image Provider using Trivy for vulnerability and secret scanning. + + This is a Tool/Wrapper provider that delegates all scanning logic to Trivy's + `trivy image` command and converts the output to Prowler's finding format. + """ + + _type: str = "image" + FINDING_BATCH_SIZE: int = 100 + MAX_IMAGE_LIST_LINES: int = 10_000 + MAX_IMAGE_NAME_LENGTH: int = 500 + _IMAGE_NAME_PATTERN = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9.\-_/:@]+$") + _SHELL_METACHARACTERS = frozenset(";|&$`\n\r") + audit_metadata: Audit_Metadata + + def __init__( + self, + images: list[str] | None = None, + image_list_file: str | None = None, + scanners: list[str] | None = None, + image_config_scanners: list[str] | None = None, + trivy_severity: list[str] | None = None, + ignore_unfixed: bool = False, + timeout: str = "5m", + config_path: str | None = None, + config_content: dict | None = None, + fixer_config: dict | None = None, + registry_username: str | None = None, + registry_password: str | None = None, + registry_token: str | None = None, + registry: str | None = None, + image_filter: str | None = None, + tag_filter: str | None = None, + max_images: int = 0, + registry_insecure: bool = False, + registry_list_images: bool = False, + ): + logger.info("Instantiating Image Provider...") + + self.images = images if images is not None else [] + self.image_list_file = image_list_file + self.scanners = scanners if scanners is not None else ["vuln", "secret"] + self.image_config_scanners = ( + image_config_scanners if image_config_scanners is not None else [] + ) + self.trivy_severity = trivy_severity if trivy_severity is not None else [] + self.ignore_unfixed = ignore_unfixed + self.timeout = timeout + self.region = "container" + self.audited_account = "image-scan" + self._session = None + self._identity = "prowler" + self._listing_only = False + + # Registry authentication (follows IaC pattern: explicit params, env vars internal) + self.registry_username = registry_username or os.environ.get( + "REGISTRY_USERNAME" + ) + self.registry_password = registry_password or os.environ.get( + "REGISTRY_PASSWORD" + ) + self.registry_token = registry_token or os.environ.get("REGISTRY_TOKEN") + + if self.registry_username and self.registry_password: + self._auth_method = "Docker login" + logger.info("Using docker login for registry authentication") + elif self.registry_token: + self._auth_method = "Registry token" + logger.info("Using registry token for authentication") + else: + self._auth_method = "No auth" + + # Registry scan mode + self.registry = registry + self.image_filter = image_filter + self.tag_filter = tag_filter + self.max_images = max_images + self.registry_insecure = registry_insecure + self.registry_list_images = registry_list_images + + # Compile regex filters + self._image_filter_re = None + self._tag_filter_re = None + if self.image_filter: + try: + self._image_filter_re = re.compile(self.image_filter) + except re.error as exc: + raise ImageInvalidFilterError( + file=__file__, + message=f"Invalid --image-filter regex '{self.image_filter}': {exc}", + ) + if self.tag_filter: + try: + self._tag_filter_re = re.compile(self.tag_filter) + except re.error as exc: + raise ImageInvalidFilterError( + file=__file__, + message=f"Invalid --tag-filter regex '{self.tag_filter}': {exc}", + ) + + self._validate_inputs() + + # Load images from file if provided + if image_list_file: + self._load_images_from_file(image_list_file) + + # Registry scan mode: enumerate images from registry + if self.registry: + self._enumerate_registry() + if self._listing_only: + return + + for image in self.images: + self._validate_image_name(image) + + if not self.images: + raise ImageNoImagesProvidedError( + file=__file__, + message="No images provided for scanning.", + ) + + # Audit Config + if config_content: + self._audit_config = config_content + else: + if not config_path: + config_path = default_config_file_path + self._audit_config = load_and_validate_config_file(self._type, config_path) + + # Fixer Config + self._fixer_config = fixer_config if fixer_config is not None else {} + + # Mutelist (not needed for Image provider since Trivy has its own logic) + self._mutelist = None + + self.audit_metadata = Audit_Metadata( + provider=self._type, + account_id=self.audited_account, + account_name="image", + region=self.region, + services_scanned=0, + expected_checks=[], + completed_checks=0, + audit_progress=0, + ) + + Provider.set_global_provider(self) + + def _load_images_from_file(self, file_path: str) -> None: + """Load image names from a file (one per line).""" + try: + line_count = 0 + with open(file_path, "r") as f: + for line in f: + line_count += 1 + if line_count > self.MAX_IMAGE_LIST_LINES: + raise ImageListFileReadError( + file=file_path, + message=f"Image list file exceeds maximum of {self.MAX_IMAGE_LIST_LINES} lines.", + ) + line = line.strip() + if not line or line.startswith("#"): + continue + if len(line) > self.MAX_IMAGE_NAME_LENGTH: + logger.warning( + f"Skipping image name exceeding {self.MAX_IMAGE_NAME_LENGTH} chars at line {line_count} in {file_path}" + ) + continue + self.images.append(line) + logger.info(f"Loaded {len(self.images)} images from {file_path}") + except FileNotFoundError: + raise ImageListFileNotFoundError( + file=file_path, + message=f"Image list file not found: {file_path}", + ) + except (ImageListFileReadError, ImageListFileNotFoundError): + raise + except Exception as error: + raise ImageListFileReadError( + file=file_path, + original_exception=error, + message=f"Error reading image list file: {error}", + ) + + def _validate_inputs(self) -> None: + """Validate timeout, scanners, and severity inputs.""" + if not re.fullmatch(r"\d+[smh]", self.timeout): + raise ImageInvalidTimeoutError( + file=__file__, + message=f"Invalid timeout format: '{self.timeout}'. Expected pattern like '5m', '300s', or '1h'.", + ) + + for scanner in self.scanners: + if scanner not in SCANNERS_CHOICES: + raise ImageInvalidScannerError( + file=__file__, + message=f"Invalid scanner: '{scanner}'. Valid options: {', '.join(SCANNERS_CHOICES)}.", + ) + + for config_scanner in self.image_config_scanners: + if config_scanner not in IMAGE_CONFIG_SCANNERS_CHOICES: + raise ImageInvalidConfigScannerError( + file=__file__, + message=f"Invalid image config scanner: '{config_scanner}'. Valid options: {', '.join(IMAGE_CONFIG_SCANNERS_CHOICES)}.", + ) + + for severity in self.trivy_severity: + if severity not in SEVERITY_CHOICES: + raise ImageInvalidSeverityError( + file=__file__, + message=f"Invalid severity: '{severity}'. Valid options: {', '.join(SEVERITY_CHOICES)}.", + ) + + def _validate_image_name(self, name: str) -> None: + """Validate a container image name for safety and correctness.""" + if not name: + raise ImageInvalidNameError( + file=__file__, + message="Image name must not be empty.", + ) + + if len(name) > self.MAX_IMAGE_NAME_LENGTH: + raise ImageInvalidNameError( + file=__file__, + message=f"Image name exceeds maximum length of {self.MAX_IMAGE_NAME_LENGTH} characters: '{name[:50]}...'", + ) + + if any(c in self._SHELL_METACHARACTERS for c in name): + raise ImageInvalidNameError( + file=__file__, + message=f"Image name contains invalid characters: '{name}'", + ) + + if not self._IMAGE_NAME_PATTERN.fullmatch(name): + raise ImageInvalidNameError( + file=__file__, + message=f"Image name does not match valid OCI reference format: '{name}'", + ) + + @property + def auth_method(self) -> str: + return self._auth_method + + @property + def type(self) -> str: + return self._type + + @property + def identity(self) -> str: + return self._identity + + @property + def session(self) -> None: + return self._session + + @property + def audit_config(self) -> dict: + return self._audit_config + + @property + def fixer_config(self) -> dict: + return self._fixer_config + + def setup_session(self) -> None: + """Image provider doesn't need a session since it uses Trivy directly""" + return None + + @staticmethod + def _extract_registry(image: str) -> str | None: + """Extract registry hostname from an image reference. + + Returns None for Docker Hub images (no registry prefix). + """ + parts = image.split("/") + if len(parts) >= 2 and ("." in parts[0] or ":" in parts[0]): + return parts[0] + return None + + def cleanup(self) -> None: + """Clean up any resources after scanning.""" + + def _process_finding( + self, + finding: dict, + image: str, + trivy_target: str, + image_sha: str = "", + ) -> CheckReportImage: + """ + Process a single finding and create a CheckReportImage object. + + Args: + finding: The finding object from Trivy output + image: The clean container image name (e.g., "alpine:3.18") + trivy_target: The Trivy target string (e.g., "alpine:3.18 (alpine 3.18.0)") + image_sha: Short SHA from Trivy Metadata.ImageID for resource uniqueness + + Returns: + CheckReportImage: The processed check report + """ + try: + # Determine finding ID and category based on type + if "VulnerabilityID" in finding: + finding_id = finding["VulnerabilityID"] + finding_description = finding.get( + "Description", finding.get("Title", "") + ) + finding_status = "FAIL" + finding_categories = ["vulnerability"] + elif "RuleID" in finding: + # Secret finding + finding_id = finding["RuleID"] + finding_description = finding.get("Title", "Secret detected") + finding_status = "FAIL" + finding_categories = ["secrets"] + else: + finding_id = finding.get("ID", "UNKNOWN") + finding_description = finding.get("Description", "") + finding_status = finding.get("Status", "FAIL") + finding_categories = [] + + # Build remediation text for vulnerabilities + remediation_text = "" + if finding.get("FixedVersion"): + remediation_text = f"Upgrade {finding.get('PkgName', 'package')} to version {finding['FixedVersion']}" + elif finding.get("Resolution"): + remediation_text = finding["Resolution"] + + # Convert Trivy severity to Prowler severity (lowercase, map UNKNOWN to informational) + trivy_severity = finding.get("Severity", "UNKNOWN").lower() + if trivy_severity == "unknown": + trivy_severity = "informational" + + metadata_dict = { + "Provider": "image", + "CheckID": finding_id, + "CheckTitle": finding.get("Title", finding_id), + "CheckType": ["Container Image Security"], + "ServiceName": "container-image", + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": trivy_severity, + "ResourceType": "container-image", + "ResourceGroup": "container", + "Description": finding_description, + "Risk": finding.get( + "Description", "Vulnerability detected in container image" + ), + "RelatedUrl": "", + "Remediation": { + "Code": { + "NativeIaC": "", + "Terraform": "", + "CLI": "", + "Other": "", + }, + "Recommendation": { + "Text": remediation_text, + "Url": finding.get("PrimaryURL", ""), + }, + }, + "Categories": finding_categories, + "DependsOn": [], + "RelatedTo": [], + "Notes": "", + } + + # Convert metadata dict to JSON string + metadata = json.dumps(metadata_dict) + + report = CheckReportImage( + metadata=metadata, finding=finding, image_name=image + ) + report.status = finding_status + report.status_extended = self._build_status_extended(finding) + report.region = self.region + report.image_sha = image_sha + report.resource_details = trivy_target + return report + + except Exception as error: + raise ImageFindingProcessingError( + file=__file__, + original_exception=error, + message=f"Error processing finding: {error}", + ) + + def _build_status_extended(self, finding: dict) -> str: + """Build a detailed status message for the finding.""" + parts = [] + + if finding.get("VulnerabilityID"): + parts.append(f"{finding['VulnerabilityID']}") + + if finding.get("PkgName"): + pkg_info = finding["PkgName"] + if finding.get("InstalledVersion"): + pkg_info += f"@{finding['InstalledVersion']}" + parts.append(f"in package {pkg_info}") + + if finding.get("FixedVersion"): + parts.append(f"(fix available: {finding['FixedVersion']})") + elif finding.get("Status") == "will_not_fix": + parts.append("(no fix available)") + + if finding.get("Title"): + parts.append(f"- {finding['Title']}") + + return ( + " ".join(parts) if parts else finding.get("Description", "Finding detected") + ) + + def run(self) -> list[CheckReportImage]: + """Execute the container image scan.""" + try: + reports = [] + for batch in self.run_scan(): + reports.extend(batch) + return reports + finally: + self.cleanup() + + def scan_per_image( + self, + ) -> Generator[tuple[str, list[CheckReportImage]], None, None]: + """Scan images one by one, yielding (image_name, findings) per image. + + Unlike run() which returns all findings at once, this method yields + after each image completes, enabling progress tracking. + """ + try: + for image in self.images: + try: + image_findings = [] + for batch in self._scan_single_image(image): + image_findings.extend(batch) + yield (image, image_findings) + except (ImageScanError, ImageTrivyBinaryNotFoundError): + raise + except Exception as error: + logger.error(f"Error scanning image {image}: {error}") + continue + finally: + self.cleanup() + + def run_scan(self) -> Generator[list[CheckReportImage], None, None]: + """ + Run Trivy scan on all configured images. + + Yields: + list[CheckReportImage]: Batches of findings + """ + for image in self.images: + try: + yield from self._scan_single_image(image) + except (ImageScanError, ImageTrivyBinaryNotFoundError): + raise + except Exception as error: + logger.error(f"Error scanning image {image}: {error}") + continue + + def _scan_single_image( + self, image: str + ) -> Generator[list[CheckReportImage], None, None]: + """ + Scan a single container image with Trivy. + + Args: + image: The container image name/tag to scan + + Yields: + list[CheckReportImage]: Batches of findings + """ + try: + logger.info(f"Scanning container image: {image}") + + # Build Trivy command + trivy_command = [ + "trivy", + "image", + "--format", + "json", + "--scanners", + ",".join(self.scanners), + "--timeout", + self.timeout, + ] + + if self.image_config_scanners: + trivy_command.extend( + ["--image-config-scanners", ",".join(self.image_config_scanners)] + ) + + if self.trivy_severity: + trivy_command.extend(["--severity", ",".join(self.trivy_severity)]) + + if self.ignore_unfixed: + trivy_command.append("--ignore-unfixed") + + trivy_command.append(image) + + # Execute Trivy + process = self._execute_trivy(trivy_command, image) + + # Log stderr output + if process.stderr: + self._log_trivy_stderr(process.stderr) + + # Check for Trivy failure + if process.returncode != 0: + error_msg = self._extract_trivy_errors(process.stderr) + categorized_msg = self._categorize_trivy_error(error_msg) + raise ImageScanError( + file=__file__, + message=f"Trivy scan failed for {image}: {categorized_msg}", + ) + + # Parse JSON output + try: + output = json.loads(process.stdout) + results = output.get("Results", []) + + if not results: + logger.info(f"No findings for image: {image}") + return + + # Extract image digest for resource uniqueness + trivy_metadata = output.get("Metadata", {}) + image_id = trivy_metadata.get("ImageID", "") + if not image_id: + repo_digests = trivy_metadata.get("RepoDigests", []) + if repo_digests: + image_id = ( + repo_digests[0].split("@")[-1] + if "@" in repo_digests[0] + else "" + ) + short_sha = image_id.replace("sha256:", "")[:12] if image_id else "" + + except json.JSONDecodeError as error: + logger.error(f"Failed to parse Trivy output for {image}: {error}") + logger.debug(f"Trivy stdout: {process.stdout[:500]}") + return + + # Process findings in batches + batch = [] + + for result in results: + target = result.get("Target", image) + + # Process Vulnerabilities + for vuln in result.get("Vulnerabilities", []): + report = self._process_finding( + vuln, image, target, image_sha=short_sha + ) + batch.append(report) + if len(batch) >= self.FINDING_BATCH_SIZE: + yield batch + batch = [] + + # Process Secrets + for secret in result.get("Secrets", []): + report = self._process_finding( + secret, image, target, image_sha=short_sha + ) + batch.append(report) + if len(batch) >= self.FINDING_BATCH_SIZE: + yield batch + batch = [] + + # Process Misconfigurations (from Dockerfile) + for misconfig in result.get("Misconfigurations", []): + report = self._process_finding( + misconfig, image, target, image_sha=short_sha + ) + batch.append(report) + if len(batch) >= self.FINDING_BATCH_SIZE: + yield batch + batch = [] + + # Yield remaining findings + if batch: + yield batch + + except (ImageScanError, ImageTrivyBinaryNotFoundError): + raise + except Exception as error: + if "No such file or directory: 'trivy'" in str(error): + raise ImageTrivyBinaryNotFoundError( + file=__file__, + original_exception=error, + message="Trivy binary not found. Please install Trivy from https://trivy.dev/latest/getting-started/installation/", + ) + 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( + ctrl_c=False, + bar="blocks", + spinner="classic", + stats=False, + enrich_print=False, + ) as bar: + bar.title = f"-> Scanning {image}..." + process = subprocess.run( + command, + capture_output=True, + text=True, + env=env, + ) + bar.title = f"-> Scan completed for {image}" + return process + else: + logger.info(f"Scanning {image}...") + process = subprocess.run( + 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, env=env) + + def _log_trivy_stderr(self, stderr: str) -> None: + """Parse and log Trivy's stderr output.""" + for line in stderr.strip().split("\n"): + if line.strip(): + parts = line.split() + if len(parts) >= 3: + level = parts[1] + message = " ".join(parts[2:]) + if level == "ERROR": + logger.error(message) + elif level == "WARN": + logger.warning(message) + elif level == "INFO": + logger.info(message) + elif level == "DEBUG": + logger.debug(message) + else: + logger.info(message) + else: + logger.info(line) + + @staticmethod + def _extract_trivy_errors(stderr: str) -> str: + """Extract only ERROR-level messages from Trivy stderr output.""" + if not stderr: + return "Unknown error" + error_lines = [] + for line in stderr.strip().split("\n"): + parts = line.split() + if len(parts) >= 3 and parts[1] == "ERROR": + error_lines.append(" ".join(parts[2:])) + elif len(parts) >= 3 and parts[1] == "FATAL": + error_lines.append(" ".join(parts[2:])) + if error_lines: + return "; ".join(error_lines)[:500] + # Fallback: no ERROR lines found, return last non-empty line + for line in reversed(stderr.strip().split("\n")): + if line.strip(): + return line.strip()[:500] + return "Unknown error" + + @staticmethod + def _categorize_trivy_error(error_msg: str) -> str: + """Categorize a Trivy error message to provide actionable guidance.""" + lower = error_msg.lower() + + if any(kw in lower for kw in ("401", "403", "unauthorized", "denied")): + return f"Auth failure — check `docker login`: {error_msg}" + if any(kw in lower for kw in ("404", "manifest unknown", "not found")): + return f"Image not found — check name/tag/registry: {error_msg}" + if any(kw in lower for kw in ("429", "rate limit", "too many requests")): + return f"Rate limited — wait or authenticate: {error_msg}" + if any(kw in lower for kw in ("timeout", "connection refused", "no such host")): + return f"Network issue — check connectivity: {error_msg}" + + return error_msg + + def _enumerate_registry(self) -> None: + """Enumerate images from a registry using the appropriate adapter.""" + verify_ssl = not self.registry_insecure + adapter = create_registry_adapter( + registry_url=self.registry, + username=self.registry_username, + password=self.registry_password, + token=self.registry_token, + verify_ssl=verify_ssl, + ) + + repositories = adapter.list_repositories() + logger.info( + f"Discovered {len(repositories)} repositories from registry {self.registry}" + ) + + # Apply image filter + if self._image_filter_re: + repositories = [r for r in repositories if self._image_filter_re.search(r)] + logger.info( + f"{len(repositories)} repositories match --image-filter '{self.image_filter}'" + ) + + if not repositories: + logger.warning( + f"No repositories found in registry {self.registry} (after filtering)" + ) + return + + # Determine if this is a Docker Hub adapter (for image reference format) + is_dockerhub = isinstance(adapter, DockerHubAdapter) + + discovered_images = [] + repos_tags: dict[str, list[str]] = {} + for repo in repositories: + tags = adapter.list_tags(repo) + + # Apply tag filter + if self._tag_filter_re: + tags = [t for t in tags if self._tag_filter_re.search(t)] + + if tags: + repos_tags[repo] = tags + + for tag in tags: + if is_dockerhub: + # Docker Hub images don't need a host prefix + image_ref = f"{repo}:{tag}" + else: + # OCI registries need the full host/repo:tag reference + registry_host = self.registry.rstrip("/") + for prefix in ("https://", "http://"): + if registry_host.startswith(prefix): + registry_host = registry_host[len(prefix) :] + break + image_ref = f"{registry_host}/{repo}:{tag}" + discovered_images.append(image_ref) + + # Registry list mode: print listing and return early + if self.registry_list_images: + self._print_registry_listing(repos_tags, len(discovered_images)) + self._listing_only = True + return + + # Check max-images limit + if self.max_images and len(discovered_images) > self.max_images: + raise ImageMaxImagesExceededError( + file=__file__, + message=f"Discovered {len(discovered_images)} images, exceeding --max-images {self.max_images}. Use --image-filter or --tag-filter to narrow results.", + ) + + # Deduplicate with explicit images + existing = set(self.images) + for img in discovered_images: + if img not in existing: + self.images.append(img) + existing.add(img) + + logger.info( + f"Discovered {len(discovered_images)} images from registry {self.registry} " + f"({len(repositories)} repositories). Total images to scan: {len(self.images)}" + ) + + def _print_registry_listing( + self, repos_tags: dict[str, list[str]], total_images: int + ) -> None: + """Print a structured listing of registry repositories and tags.""" + num_repos = len(repos_tags) + print( + f"\n{Style.BRIGHT}Registry:{Style.RESET_ALL} " + f"{Fore.CYAN}{self.registry}{Style.RESET_ALL} " + f"({num_repos} {'repository' if num_repos == 1 else 'repositories'}, " + f"{total_images} {'image' if total_images == 1 else 'images'})\n" + ) + for repo, tags in repos_tags.items(): + print(f" {Fore.YELLOW}{repo}{Style.RESET_ALL} " f"({len(tags)} tags)") + print(f" {', '.join(tags)}") + print() + + def print_credentials(self) -> None: + """Print scan configuration.""" + report_title = f"{Style.BRIGHT}Scanning container images:{Style.RESET_ALL}" + + report_lines = [] + if len(self.images) <= 3: + for img in self.images: + report_lines.append(f"Image: {Fore.YELLOW}{img}{Style.RESET_ALL}") + else: + report_lines.append( + f"Images: {Fore.YELLOW}{len(self.images)} images{Style.RESET_ALL}" + ) + + report_lines.append( + f"Scanners: {Fore.YELLOW}{', '.join(self.scanners)}{Style.RESET_ALL}" + ) + + if self.image_config_scanners: + report_lines.append( + f"Image config scanners: {Fore.YELLOW}{', '.join(self.image_config_scanners)}{Style.RESET_ALL}" + ) + + if self.trivy_severity: + report_lines.append( + f"Severity filter: {Fore.YELLOW}{', '.join(self.trivy_severity)}{Style.RESET_ALL}" + ) + + if self.ignore_unfixed: + report_lines.append(f"Ignore unfixed: {Fore.YELLOW}Yes{Style.RESET_ALL}") + + 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}" + ) + + if self.registry: + report_lines.append( + f"Registry: {Fore.YELLOW}{self.registry}{Style.RESET_ALL}" + ) + if self.image_filter: + report_lines.append( + f"Image filter: {Fore.YELLOW}{self.image_filter}{Style.RESET_ALL}" + ) + if self.tag_filter: + report_lines.append( + f"Tag filter: {Fore.YELLOW}{self.tag_filter}{Style.RESET_ALL}" + ) + + print_boxes(report_lines, report_title) + + @staticmethod + def test_connection( + image: str | None = None, + raise_on_exception: bool = True, + provider_id: str | None = None, + registry_username: str | None = None, + registry_password: str | None = None, + registry_token: str | None = None, + ) -> "Connection": + """ + Test connection to container registry by attempting to inspect an image. + + Args: + 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 + """ + try: + if provider_id and not image: + image = provider_id + + 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( + [ + "trivy", + "image", + "--skip-db-update", + "--download-db-only=false", + image, + ], + capture_output=True, + text=True, + timeout=60, + env=env, + ) + + if process.returncode == 0: + return Connection(is_connected=True) + else: + error_msg = process.stderr or "Unknown error" + if "401" in error_msg or "unauthorized" in error_msg.lower(): + return Connection( + is_connected=False, + error="Authentication failed. Check registry credentials.", + ) + elif "not found" in error_msg.lower() or "404" in error_msg: + return Connection( + is_connected=False, + error="Image not found in registry.", + ) + else: + return Connection( + is_connected=False, + error=f"Failed to access image: {error_msg[:200]}", + ) + + except subprocess.TimeoutExpired: + return Connection( + is_connected=False, + error="Connection timed out", + ) + except FileNotFoundError: + return Connection( + is_connected=False, + error="Trivy binary not found. Please install Trivy.", + ) + except Exception as error: + if raise_on_exception: + raise + return Connection( + is_connected=False, + error=f"Unexpected error: {str(error)}", + ) diff --git a/prowler/providers/image/lib/__init__.py b/prowler/providers/image/lib/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/image/lib/arguments/__init__.py b/prowler/providers/image/lib/arguments/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/image/lib/arguments/arguments.py b/prowler/providers/image/lib/arguments/arguments.py new file mode 100644 index 0000000000..3dfe9cf92b --- /dev/null +++ b/prowler/providers/image/lib/arguments/arguments.py @@ -0,0 +1,183 @@ +SCANNERS_CHOICES = [ + "vuln", + "secret", + "misconfig", + "license", +] + +IMAGE_CONFIG_SCANNERS_CHOICES = [ + "misconfig", + "secret", +] + +SEVERITY_CHOICES = [ + "CRITICAL", + "HIGH", + "MEDIUM", + "LOW", + "UNKNOWN", +] + + +def init_parser(self): + """Init the Image Provider CLI parser""" + image_parser = self.subparsers.add_parser( + "image", parents=[self.common_providers_parser], help="Container Image Provider" + ) + + # Image Selection + image_selection_group = image_parser.add_argument_group("Image Selection") + image_selection_group.add_argument( + "--image", + "-I", + dest="images", + action="append", + default=[], + help="Container image to scan. Can be specified multiple times. Examples: nginx:latest, alpine:3.18, myregistry.io/myapp:v1.0", + ) + + image_selection_group.add_argument( + "--image-list", + dest="image_list_file", + default=None, + help="Path to a file containing list of images to scan (one per line). Lines starting with # are treated as comments.", + ) + + # Scan Configuration + scan_config_group = image_parser.add_argument_group("Scan Configuration") + scan_config_group.add_argument( + "--scanners", + "--scanner", + dest="scanners", + nargs="+", + default=["vuln", "secret"], + choices=SCANNERS_CHOICES, + help="Trivy scanners to use. Default: vuln, secret. Available: vuln, secret, misconfig, license", + ) + + scan_config_group.add_argument( + "--image-config-scanners", + dest="image_config_scanners", + nargs="+", + default=[], + choices=IMAGE_CONFIG_SCANNERS_CHOICES, + help="Trivy image config scanners (scans Dockerfile-level metadata). Available: misconfig, secret", + ) + + scan_config_group.add_argument( + "--trivy-severity", + dest="trivy_severity", + nargs="+", + default=[], + choices=SEVERITY_CHOICES, + help="Filter Trivy findings by severity. Default: all severities. Available: CRITICAL, HIGH, MEDIUM, LOW, UNKNOWN", + ) + + scan_config_group.add_argument( + "--ignore-unfixed", + dest="ignore_unfixed", + action="store_true", + default=False, + help="Ignore vulnerabilities without available fixes.", + ) + + scan_config_group.add_argument( + "--timeout", + dest="timeout", + default="5m", + help="Trivy scan timeout. Default: 5m. Examples: 10m, 1h", + ) + + # Registry Scan Mode + registry_group = image_parser.add_argument_group("Registry Scan Mode") + registry_group.add_argument( + "--registry", + dest="registry", + default=None, + help="Registry URL to enumerate and scan all images. Examples: myregistry.io, docker.io/myorg, 123456789.dkr.ecr.us-east-1.amazonaws.com", + ) + registry_group.add_argument( + "--image-filter", + dest="image_filter", + default=None, + help="Regex to filter repository names during registry enumeration (re.search). Example: '^prod/.*'", + ) + registry_group.add_argument( + "--tag-filter", + dest="tag_filter", + default=None, + help=r"Regex to filter tags during registry enumeration (re.search). Example: '^(latest|v\d+\.\d+\.\d+)$'", + ) + registry_group.add_argument( + "--max-images", + dest="max_images", + type=int, + default=0, + help="Maximum number of images to scan from registry. 0 = unlimited. Aborts if exceeded.", + ) + registry_group.add_argument( + "--registry-insecure", + dest="registry_insecure", + action="store_true", + default=False, + help="Skip TLS verification for registry connections (for self-signed certificates).", + ) + registry_group.add_argument( + "--registry-list", + dest="registry_list_images", + action="store_true", + default=False, + help="List all repositories and tags from the registry, then exit without scanning. Useful for discovering available images before building --image-filter or --tag-filter.", + ) + + +def validate_arguments(arguments): + """Validate Image provider arguments.""" + images = getattr(arguments, "images", []) + image_list_file = getattr(arguments, "image_list_file", None) + registry = getattr(arguments, "registry", None) + image_filter = getattr(arguments, "image_filter", None) + tag_filter = getattr(arguments, "tag_filter", None) + max_images = getattr(arguments, "max_images", 0) + registry_insecure = getattr(arguments, "registry_insecure", False) + registry_list_images = getattr(arguments, "registry_list_images", False) + + if registry_list_images and not registry: + return (False, "--registry-list requires --registry.") + + if not images and not image_list_file and not registry: + return ( + False, + "At least one image source must be specified using --image (-I), --image-list, or --registry.", + ) + + # Registry-only flags require --registry + if not registry: + if image_filter: + return (False, "--image-filter requires --registry.") + if tag_filter: + return (False, "--tag-filter requires --registry.") + if max_images: + return (False, "--max-images requires --registry.") + if registry_insecure: + return (False, "--registry-insecure requires --registry.") + + # Docker Hub namespace validation + if registry: + url = registry.rstrip("/") + for prefix in ("https://", "http://"): + if url.startswith(prefix): + url = url[len(prefix) :] + break + stripped = url + for prefix in ("registry-1.docker.io", "docker.io"): + if stripped.startswith(prefix): + stripped = stripped[len(prefix) :].lstrip("/") + if not stripped: + return ( + False, + "Docker Hub requires a namespace. Use --registry docker.io/{org_or_user}.", + ) + break + + return (True, "") diff --git a/prowler/providers/image/lib/registry/__init__.py b/prowler/providers/image/lib/registry/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/image/lib/registry/base.py b/prowler/providers/image/lib/registry/base.py new file mode 100644 index 0000000000..133ef9d4e0 --- /dev/null +++ b/prowler/providers/image/lib/registry/base.py @@ -0,0 +1,141 @@ +"""Registry adapter abstract base class.""" + +from __future__ import annotations + +import re +import time +from abc import ABC, abstractmethod + +import requests + +from prowler.config.config import prowler_version +from prowler.lib.logger import logger +from prowler.providers.image.exceptions.exceptions import ImageRegistryNetworkError + +_MAX_RETRIES = 3 +_BACKOFF_BASE = 1 +_USER_AGENT = f"Prowler/{prowler_version} (registry-adapter)" + + +class RegistryAdapter(ABC): + """Abstract base class for registry adapters.""" + + def __init__( + self, + registry_url: str, + username: str | None = None, + password: str | None = None, + token: str | None = None, + verify_ssl: bool = True, + ) -> None: + self.registry_url = registry_url + self.username = username + self._password = password + self._token = token + self.verify_ssl = verify_ssl + + @property + def password(self) -> str | None: + return self._password + + @property + def token(self) -> str | None: + return self._token + + def __getstate__(self) -> dict: + state = self.__dict__.copy() + state["_password"] = "***" if state.get("_password") else None + state["_token"] = "***" if state.get("_token") else None + return state + + def __repr__(self) -> str: + return ( + f"{self.__class__.__name__}(" + f"registry_url={self.registry_url!r}, " + f"username={self.username!r}, " + f"password={'' if self._password else None}, " + f"token={'' if self._token else None})" + ) + + @abstractmethod + def list_repositories(self) -> list[str]: + """Enumerate all repository names in the registry.""" + ... + + @abstractmethod + def list_tags(self, repository: str) -> list[str]: + """Enumerate all tags for a repository.""" + ... + + def _request_with_retry(self, method: str, url: str, **kwargs) -> requests.Response: + context_label = kwargs.pop("context_label", None) or self.registry_url + kwargs.setdefault("timeout", 30) + kwargs.setdefault("verify", self.verify_ssl) + headers = kwargs.get("headers", {}) + headers.setdefault("User-Agent", _USER_AGENT) + kwargs["headers"] = headers + last_exception = None + last_status = None + last_body = None + for attempt in range(1, _MAX_RETRIES + 1): + try: + resp = requests.request(method, url, **kwargs) + if resp.status_code == 429: + last_status = 429 + wait = _BACKOFF_BASE * (2 ** (attempt - 1)) + logger.warning( + f"Rate limited by {context_label}, retrying in {wait}s (attempt {attempt}/{_MAX_RETRIES})" + ) + time.sleep(wait) + continue + if resp.status_code >= 500: + last_status = resp.status_code + last_body = (resp.text or "")[:500] + wait = _BACKOFF_BASE * (2 ** (attempt - 1)) + logger.warning( + f"Server error from {context_label} (HTTP {resp.status_code}), " + f"retrying in {wait}s (attempt {attempt}/{_MAX_RETRIES}): {last_body}" + ) + time.sleep(wait) + continue + return resp + except requests.exceptions.ConnectionError as exc: + last_exception = exc + if attempt < _MAX_RETRIES: + wait = _BACKOFF_BASE * (2 ** (attempt - 1)) + logger.warning( + f"Connection error to {context_label}, retrying in {wait}s (attempt {attempt}/{_MAX_RETRIES})" + ) + time.sleep(wait) + continue + except requests.exceptions.Timeout as exc: + raise ImageRegistryNetworkError( + file=__file__, + message=f"Connection timed out to {context_label}.", + original_exception=exc, + ) + if last_status == 429: + raise ImageRegistryNetworkError( + file=__file__, + message=f"Rate limited by {context_label} after {_MAX_RETRIES} attempts.", + ) + if last_status is not None and last_status >= 500: + raise ImageRegistryNetworkError( + file=__file__, + message=f"Server error from {context_label} (HTTP {last_status}) after {_MAX_RETRIES} attempts: {last_body}", + ) + raise ImageRegistryNetworkError( + file=__file__, + message=f"Failed to connect to {context_label} after {_MAX_RETRIES} attempts.", + original_exception=last_exception, + ) + + @staticmethod + def _next_page_url(resp: requests.Response) -> str | None: + link_header = resp.headers.get("Link", "") + if not link_header: + return None + match = re.search(r'<([^>]+)>;\s*rel="next"', link_header) + if match: + return match.group(1) + return None diff --git a/prowler/providers/image/lib/registry/dockerhub_adapter.py b/prowler/providers/image/lib/registry/dockerhub_adapter.py new file mode 100644 index 0000000000..949de2d9cb --- /dev/null +++ b/prowler/providers/image/lib/registry/dockerhub_adapter.py @@ -0,0 +1,221 @@ +"""Docker Hub registry adapter.""" + +from __future__ import annotations + +import re +from typing import TYPE_CHECKING + +from prowler.lib.logger import logger +from prowler.providers.image.exceptions.exceptions import ( + ImageRegistryAuthError, + ImageRegistryCatalogError, + ImageRegistryNetworkError, +) +from prowler.providers.image.lib.registry.base import RegistryAdapter + +if TYPE_CHECKING: + import requests + +_HUB_API = "https://hub.docker.com" +_REGISTRY_HOST = "https://registry-1.docker.io" +_AUTH_URL = "https://auth.docker.io/token" + + +class DockerHubAdapter(RegistryAdapter): + """Adapter for Docker Hub using the Hub REST API + OCI tag listing.""" + + def __init__( + self, + registry_url: str, + username: str | None = None, + password: str | None = None, + token: str | None = None, + verify_ssl: bool = True, + ) -> None: + if not verify_ssl: + logger.warning( + "Docker Hub always uses TLS verification; --registry-insecure is ignored for Docker Hub registries." + ) + super().__init__(registry_url, username, password, token, verify_ssl=True) + self.namespace = self._extract_namespace(registry_url) + self._hub_jwt: str | None = None + self._registry_tokens: dict[str, str] = {} + + @staticmethod + def _extract_namespace(registry_url: str) -> str: + url = registry_url.rstrip("/") + for prefix in ( + "https://registry-1.docker.io", + "http://registry-1.docker.io", + "https://docker.io", + "http://docker.io", + "registry-1.docker.io", + "docker.io", + "https://", + "http://", + ): + if url.startswith(prefix): + url = url[len(prefix) :] + break + url = url.lstrip("/") + parts = url.split("/") + namespace = parts[0] if parts and parts[0] else "" + return namespace + + def list_repositories(self) -> list[str]: + if not self.namespace: + raise ImageRegistryCatalogError( + file=__file__, + message="Docker Hub requires a namespace. Use --registry docker.io/{org_or_user}.", + ) + self._hub_login() + repositories: list[str] = [] + if self._hub_jwt: + url = f"{_HUB_API}/v2/namespaces/{self.namespace}/repositories" + else: + url = f"{_HUB_API}/v2/repositories/{self.namespace}/" + params: dict = {"page_size": 100} + while url: + resp = self._hub_request("GET", url, params=params) + self._check_hub_response(resp, "repository listing") + data = resp.json() + for repo in data.get("results", []): + name = repo.get("name", "") + if name: + repositories.append(f"{self.namespace}/{name}") + url = data.get("next") + params = {} + return repositories + + def list_tags(self, repository: str) -> list[str]: + token = self._get_registry_token(repository) + tags: list[str] = [] + url = f"{_REGISTRY_HOST}/v2/{repository}/tags/list" + params: dict = {"n": 100} + while url: + resp = self._registry_request("GET", url, token, params=params) + if resp.status_code in (401, 403): + raise ImageRegistryAuthError( + file=__file__, + message=f"Authentication failed for tag listing of {repository} on Docker Hub. Check REGISTRY_USERNAME and REGISTRY_PASSWORD.", + ) + if resp.status_code != 200: + logger.warning( + f"Failed to list tags for {repository} (HTTP {resp.status_code}): {resp.text[:200]}" + ) + break + data = resp.json() + tags.extend(data.get("tags", []) or []) + url = self._next_tag_page_url(resp) + params = {} + return tags + + def _hub_login(self) -> None: + if self._hub_jwt: + return + if not self.username or not self.password: + return + logger.debug(f"Docker Hub login attempt for username: {self.username!r}") + resp = self._request_with_retry( + "POST", + f"{_HUB_API}/v2/users/login", + json={"username": self.username, "password": self.password}, + context_label="Docker Hub", + ) + if resp.status_code != 200: + body_preview = resp.text[:200] if resp.text else "(empty body)" + raise ImageRegistryAuthError( + file=__file__, + message=( + f"Docker Hub login failed (HTTP {resp.status_code}). " + f"Check REGISTRY_USERNAME and REGISTRY_PASSWORD. " + f"Response: {body_preview}" + ), + ) + self._hub_jwt = resp.json().get("token") + if not self._hub_jwt: + raise ImageRegistryAuthError( + file=__file__, + message="Docker Hub login returned an empty JWT token. Check REGISTRY_USERNAME and REGISTRY_PASSWORD.", + ) + + def _get_registry_token(self, repository: str) -> str: + if repository in self._registry_tokens: + return self._registry_tokens[repository] + params = { + "service": "registry.docker.io", + "scope": f"repository:{repository}:pull", + } + auth = None + if self.username and self.password: + auth = (self.username, self.password) + resp = self._request_with_retry( + "GET", + _AUTH_URL, + params=params, + auth=auth, + context_label="Docker Hub", + ) + if resp.status_code != 200: + raise ImageRegistryAuthError( + file=__file__, + message=f"Failed to obtain Docker Hub registry token for {repository} (HTTP {resp.status_code}). Check REGISTRY_USERNAME and REGISTRY_PASSWORD.", + ) + token = resp.json().get("token", "") + if not token: + raise ImageRegistryAuthError( + file=__file__, + message=f"Docker Hub registry token endpoint returned an empty token for {repository}. Check REGISTRY_USERNAME and REGISTRY_PASSWORD.", + ) + self._registry_tokens[repository] = token + return token + + def _hub_request(self, method: str, url: str, **kwargs) -> requests.Response: + headers = kwargs.pop("headers", {}) + if self._hub_jwt: + headers["Authorization"] = f"Bearer {self._hub_jwt}" + kwargs["headers"] = headers + return self._request_with_retry( + method, url, context_label="Docker Hub", **kwargs + ) + + def _registry_request( + self, method: str, url: str, token: str, **kwargs + ) -> requests.Response: + headers = kwargs.pop("headers", {}) + headers["Authorization"] = f"Bearer {token}" + kwargs["headers"] = headers + return self._request_with_retry( + method, url, context_label="Docker Hub", **kwargs + ) + + def _check_hub_response(self, resp: requests.Response, context: str) -> None: + if resp.status_code == 200: + return + if resp.status_code in (401, 403): + raise ImageRegistryAuthError( + file=__file__, + message=f"Authentication failed for {context} on Docker Hub (HTTP {resp.status_code}). Check REGISTRY_USERNAME and REGISTRY_PASSWORD environment variables.", + ) + if resp.status_code == 404: + raise ImageRegistryCatalogError( + file=__file__, + message=f"Namespace '{self.namespace}' not found on Docker Hub. Check the namespace in --registry docker.io/{{namespace}}.", + ) + raise ImageRegistryNetworkError( + file=__file__, + message=f"Unexpected error during {context} on Docker Hub (HTTP {resp.status_code}): {resp.text[:200]}", + ) + + @staticmethod + def _next_tag_page_url(resp: requests.Response) -> str | None: + link_header = resp.headers.get("Link", "") + if not link_header: + return None + match = re.search(r'<([^>]+)>;\s*rel="next"', link_header) + if match: + next_url = match.group(1) + if next_url.startswith("/"): + return f"{_REGISTRY_HOST}{next_url}" + return next_url + return None diff --git a/prowler/providers/image/lib/registry/factory.py b/prowler/providers/image/lib/registry/factory.py new file mode 100644 index 0000000000..5c0134fedd --- /dev/null +++ b/prowler/providers/image/lib/registry/factory.py @@ -0,0 +1,40 @@ +"""Factory for auto-detecting registry type and returning the appropriate adapter.""" + +from __future__ import annotations + +import re + +from prowler.providers.image.lib.registry.base import RegistryAdapter +from prowler.providers.image.lib.registry.dockerhub_adapter import DockerHubAdapter +from prowler.providers.image.lib.registry.oci_adapter import OciRegistryAdapter + +_DOCKER_HUB_PATTERN = re.compile( + r"^(https?://)?(docker\.io|registry-1\.docker\.io)(/|$)", re.IGNORECASE +) + + +def create_registry_adapter( + registry_url: str, + username: str | None = None, + password: str | None = None, + token: str | None = None, + verify_ssl: bool = True, +) -> RegistryAdapter: + """Auto-detect registry type from URL and return the appropriate adapter.""" + if _DOCKER_HUB_PATTERN.search(registry_url): + return DockerHubAdapter( + registry_url=registry_url, + username=username, + password=password, + token=token, + verify_ssl=verify_ssl, + ) + # ECR and other non-Docker-Hub registries implement the OCI Distribution Spec, + # so they are handled by the generic OCI adapter. + return OciRegistryAdapter( + registry_url=registry_url, + username=username, + password=password, + token=token, + verify_ssl=verify_ssl, + ) diff --git a/prowler/providers/image/lib/registry/oci_adapter.py b/prowler/providers/image/lib/registry/oci_adapter.py new file mode 100644 index 0000000000..b7eb4c780b --- /dev/null +++ b/prowler/providers/image/lib/registry/oci_adapter.py @@ -0,0 +1,228 @@ +"""Generic OCI Distribution Spec registry adapter.""" + +from __future__ import annotations + +import base64 +import ipaddress +import re +from typing import TYPE_CHECKING +from urllib.parse import urlparse + +from prowler.lib.logger import logger +from prowler.providers.image.exceptions.exceptions import ( + ImageRegistryAuthError, + ImageRegistryCatalogError, + ImageRegistryNetworkError, +) +from prowler.providers.image.lib.registry.base import RegistryAdapter + +if TYPE_CHECKING: + import requests + + +class OciRegistryAdapter(RegistryAdapter): + """Adapter for registries implementing OCI Distribution Spec.""" + + def __init__( + self, + registry_url: str, + username: str | None = None, + password: str | None = None, + token: str | None = None, + verify_ssl: bool = True, + ) -> None: + super().__init__(registry_url, username, password, token, verify_ssl) + self._base_url = self._normalise_url(registry_url) + self._bearer_token: str | None = None + self._basic_auth_verified = False + + @staticmethod + def _normalise_url(url: str) -> str: + url = url.rstrip("/") + if not url.startswith(("http://", "https://")): + url = f"https://{url}" + return url + + def list_repositories(self) -> list[str]: + self._ensure_auth() + repositories: list[str] = [] + url = f"{self._base_url}/v2/_catalog" + params: dict = {"n": 200} + while url: + resp = self._authed_request("GET", url, params=params) + if resp.status_code == 404: + raise ImageRegistryCatalogError( + file=__file__, + message=f"Registry at {self.registry_url} does not support catalog listing (/_catalog returned 404). Use --image or --image-list instead.", + ) + self._check_response(resp, "catalog listing") + data = resp.json() + repositories.extend(data.get("repositories", [])) + url = self._next_page_url(resp) + params = {} + return repositories + + def list_tags(self, repository: str) -> list[str]: + self._ensure_auth(repository=repository) + tags: list[str] = [] + url = f"{self._base_url}/v2/{repository}/tags/list" + params: dict = {"n": 200} + while url: + resp = self._authed_request("GET", url, params=params) + self._check_response(resp, f"tag listing for {repository}") + data = resp.json() + tags.extend(data.get("tags", []) or []) + url = self._next_page_url(resp) + params = {} + return tags + + def _ensure_auth(self, repository: str | None = None) -> None: + if self._bearer_token: + return + if self._basic_auth_verified: + return + if self.token: + self._bearer_token = self.token + return + ping_url = f"{self._base_url}/v2/" + resp = self._request_with_retry("GET", ping_url) + if resp.status_code == 200: + 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: + raise ImageRegistryAuthError( + file=__file__, + message=f"Access denied to registry {self.registry_url} (HTTP 403). Check REGISTRY_USERNAME and REGISTRY_PASSWORD.", + ) + raise ImageRegistryNetworkError( + file=__file__, + message=f"Unexpected HTTP {resp.status_code} from registry {self.registry_url} during auth check.", + ) + + def _obtain_bearer_token( + self, www_authenticate: str, repository: str | None = None + ) -> str: + match = re.search(r'realm="([^"]+)"', www_authenticate) + if not match: + raise ImageRegistryAuthError( + file=__file__, + message=f"Cannot parse token endpoint from registry {self.registry_url}. Www-Authenticate: {www_authenticate[:200]}", + ) + realm = match.group(1) + self._validate_realm_url(realm) + params: dict = {} + service_match = re.search(r'service="([^"]+)"', www_authenticate) + if service_match: + params["service"] = service_match.group(1) + scope_match = re.search(r'scope="([^"]+)"', www_authenticate) + if scope_match: + params["scope"] = scope_match.group(1) + elif repository: + params["scope"] = f"repository:{repository}:pull" + auth = None + if self.username and self.password: + auth = (self.username, self.password) + resp = self._request_with_retry("GET", realm, params=params, auth=auth) + if resp.status_code != 200: + raise ImageRegistryAuthError( + file=__file__, + message=f"Failed to obtain bearer token from {realm} (HTTP {resp.status_code}). Check REGISTRY_USERNAME and REGISTRY_PASSWORD.", + ) + data = resp.json() + token = data.get("token") or data.get("access_token", "") + if not token: + raise ImageRegistryAuthError( + file=__file__, + message=f"Token endpoint {realm} returned an empty token. Check REGISTRY_USERNAME and REGISTRY_PASSWORD.", + ) + return token + + @staticmethod + def _validate_realm_url(realm: str) -> None: + parsed = urlparse(realm) + if parsed.scheme not in ("http", "https"): + raise ImageRegistryAuthError( + file=__file__, + message=f"Bearer token realm has disallowed scheme: {parsed.scheme}. Only http/https are allowed.", + ) + if parsed.scheme == "http": + logger.warning(f"Bearer token realm uses HTTP (not HTTPS): {realm}") + hostname = parsed.hostname or "" + try: + addr = ipaddress.ip_address(hostname) + if addr.is_private or addr.is_loopback or addr.is_link_local: + raise ImageRegistryAuthError( + file=__file__, + message=f"Bearer token realm points to a private/loopback address: {hostname}. This may indicate an SSRF attempt.", + ) + except ValueError: + pass + + def _resolve_basic_credentials(self) -> tuple[str | None, str | None]: + """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. + """ + if not self.password: + return self.username, self.password + try: + decoded = base64.b64decode(self.password).decode("utf-8") + if decoded.startswith(f"{self.username}:"): + return self.username, decoded[len(self.username) + 1 :] + except (ValueError, UnicodeDecodeError): + logger.debug("Password is not a base64-encoded auth token, using as-is") + 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}" + elif self.username and self.password: + user, pwd = self._resolve_basic_credentials() + kwargs.setdefault("auth", (user, pwd)) + kwargs["headers"] = headers + return self._request_with_retry(method, url, **kwargs) + + def _check_response(self, resp: requests.Response, context: str) -> None: + if resp.status_code == 200: + return + if resp.status_code in (401, 403): + raise ImageRegistryAuthError( + file=__file__, + message=f"Authentication failed for {context} on {self.registry_url} (HTTP {resp.status_code}). Check REGISTRY_USERNAME and REGISTRY_PASSWORD.", + ) + raise ImageRegistryNetworkError( + file=__file__, + message=f"Unexpected error during {context} on {self.registry_url} (HTTP {resp.status_code}): {resp.text[:200]}", + ) diff --git a/prowler/providers/image/models.py b/prowler/providers/image/models.py new file mode 100644 index 0000000000..232f8a6e39 --- /dev/null +++ b/prowler/providers/image/models.py @@ -0,0 +1,21 @@ +from prowler.config.config import output_file_timestamp +from prowler.providers.common.models import ProviderOutputOptions + + +class ImageOutputOptions(ProviderOutputOptions): + """ + ImageOutputOptions customizes output filename logic for container image scanning. + + Attributes inherited from ProviderOutputOptions: + - output_filename (str): The base filename used for generated reports. + - output_directory (str): The directory to store the output files. + """ + + def __init__(self, arguments, bulk_checks_metadata): + super().__init__(arguments, bulk_checks_metadata) + + # If --output-filename is not specified, build a default name + if not getattr(arguments, "output_filename", None): + self.output_filename = f"prowler-output-image-{output_file_timestamp}" + else: + self.output_filename = arguments.output_filename diff --git a/tests/providers/image/image_fixtures.py b/tests/providers/image/image_fixtures.py new file mode 100644 index 0000000000..920a7f225a --- /dev/null +++ b/tests/providers/image/image_fixtures.py @@ -0,0 +1,143 @@ +import json + +# Sample vulnerability finding from Trivy +SAMPLE_VULNERABILITY_FINDING = { + "VulnerabilityID": "CVE-2024-1234", + "PkgID": "openssl@1.1.1k-r0", + "PkgName": "openssl", + "InstalledVersion": "1.1.1k-r0", + "FixedVersion": "1.1.1l-r0", + "Severity": "HIGH", + "Title": "OpenSSL Buffer Overflow", + "Description": "A buffer overflow vulnerability in OpenSSL allows remote attackers to execute arbitrary code.", + "PrimaryURL": "https://avd.aquasec.com/nvd/cve-2024-1234", +} + +# Sample secret finding from Trivy +SAMPLE_SECRET_FINDING = { + "RuleID": "aws-access-key-id", + "Category": "AWS", + "Severity": "CRITICAL", + "Title": "AWS Access Key ID", + "StartLine": 10, + "EndLine": 10, + "Match": "AKIA...", +} + +# Sample misconfiguration finding from Trivy +SAMPLE_MISCONFIGURATION_FINDING = { + "ID": "DS001", + "Title": "Dockerfile should not use latest tag", + "Description": "Using latest tag can cause unpredictable builds.", + "Severity": "MEDIUM", + "Resolution": "Use a specific version tag instead of latest", + "PrimaryURL": "https://avd.aquasec.com/misconfig/ds001", +} + +# Sample finding with UNKNOWN severity +SAMPLE_UNKNOWN_SEVERITY_FINDING = { + "VulnerabilityID": "CVE-2024-9999", + "PkgID": "test-pkg@0.0.1", + "PkgName": "test-pkg", + "InstalledVersion": "0.0.1", + "Severity": "UNKNOWN", + "Title": "Unknown severity issue", + "Description": "An issue with unknown severity.", +} + +# Sample image SHA for testing (first 12 chars of a sha256 digest) +SAMPLE_IMAGE_SHA = "c1aabb73d233" +SAMPLE_IMAGE_ID = f"sha256:{SAMPLE_IMAGE_SHA}abcdef1234567890" + +# Full Trivy JSON output structure with a single vulnerability +SAMPLE_TRIVY_IMAGE_OUTPUT = { + "Metadata": { + "ImageID": SAMPLE_IMAGE_ID, + "RepoDigests": [f"alpine@sha256:{SAMPLE_IMAGE_SHA}abcdef1234567890"], + }, + "Results": [ + { + "Target": "alpine:3.18 (alpine 3.18.0)", + "Type": "alpine", + "Vulnerabilities": [SAMPLE_VULNERABILITY_FINDING], + "Secrets": [], + "Misconfigurations": [], + } + ], +} + +# Full Trivy JSON output with mixed finding types +SAMPLE_TRIVY_MULTI_TYPE_OUTPUT = { + "Metadata": { + "ImageID": SAMPLE_IMAGE_ID, + "RepoDigests": [f"myimage@sha256:{SAMPLE_IMAGE_SHA}abcdef1234567890"], + }, + "Results": [ + { + "Target": "myimage:latest (debian 12)", + "Type": "debian", + "Vulnerabilities": [SAMPLE_VULNERABILITY_FINDING], + "Secrets": [SAMPLE_SECRET_FINDING], + "Misconfigurations": [SAMPLE_MISCONFIGURATION_FINDING], + } + ], +} + +# Trivy output with only RepoDigests (no ImageID) for fallback testing +SAMPLE_TRIVY_REPO_DIGEST_ONLY_OUTPUT = { + "Metadata": { + "RepoDigests": ["alpine@sha256:e5f6g7h8i9j0abcdef1234567890"], + }, + "Results": [ + { + "Target": "alpine:3.18 (alpine 3.18.0)", + "Type": "alpine", + "Vulnerabilities": [SAMPLE_VULNERABILITY_FINDING], + "Secrets": [], + "Misconfigurations": [], + } + ], +} + +# Trivy output with no Metadata at all +SAMPLE_TRIVY_NO_METADATA_OUTPUT = { + "Results": [ + { + "Target": "alpine:3.18 (alpine 3.18.0)", + "Type": "alpine", + "Vulnerabilities": [SAMPLE_VULNERABILITY_FINDING], + "Secrets": [], + "Misconfigurations": [], + } + ], +} + + +def get_sample_trivy_json_output(): + """Return sample Trivy JSON output as string.""" + return json.dumps(SAMPLE_TRIVY_IMAGE_OUTPUT) + + +def get_empty_trivy_output(): + """Return empty Trivy output as string.""" + return json.dumps({"Results": []}) + + +def get_invalid_trivy_output(): + """Return invalid JSON output as string.""" + return "invalid json output" + + +def get_multi_type_trivy_output(): + """Return Trivy output with multiple finding types as string.""" + return json.dumps(SAMPLE_TRIVY_MULTI_TYPE_OUTPUT) + + +def get_repo_digest_only_trivy_output(): + """Return Trivy output with only RepoDigests (no ImageID) as string.""" + return json.dumps(SAMPLE_TRIVY_REPO_DIGEST_ONLY_OUTPUT) + + +def get_no_metadata_trivy_output(): + """Return Trivy output with no Metadata as string.""" + return json.dumps(SAMPLE_TRIVY_NO_METADATA_OUTPUT) diff --git a/tests/providers/image/image_provider_test.py b/tests/providers/image/image_provider_test.py new file mode 100644 index 0000000000..fc36b9e617 --- /dev/null +++ b/tests/providers/image/image_provider_test.py @@ -0,0 +1,930 @@ +import os +import tempfile +from unittest import mock +from unittest.mock import MagicMock, patch + +import pytest + +from prowler.lib.check.models import CheckReportImage +from prowler.providers.image.exceptions.exceptions import ( + ImageInvalidConfigScannerError, + ImageInvalidNameError, + ImageInvalidScannerError, + ImageInvalidSeverityError, + ImageInvalidTimeoutError, + ImageListFileNotFoundError, + ImageListFileReadError, + ImageNoImagesProvidedError, + ImageScanError, + ImageTrivyBinaryNotFoundError, +) +from prowler.providers.image.image_provider import ImageProvider +from tests.providers.image.image_fixtures import ( + SAMPLE_IMAGE_SHA, + SAMPLE_MISCONFIGURATION_FINDING, + SAMPLE_SECRET_FINDING, + SAMPLE_UNKNOWN_SEVERITY_FINDING, + SAMPLE_VULNERABILITY_FINDING, + get_empty_trivy_output, + get_invalid_trivy_output, + get_multi_type_trivy_output, + get_no_metadata_trivy_output, + get_repo_digest_only_trivy_output, + get_sample_trivy_json_output, +) + + +def _make_provider(**kwargs): + """Helper to create an ImageProvider with test defaults.""" + defaults = { + "images": ["alpine:3.18"], + "config_content": {}, + } + defaults.update(kwargs) + return ImageProvider(**defaults) + + +class TestImageProvider: + def test_image_provider(self): + """Test default initialization.""" + provider = _make_provider() + + assert provider._type == "image" + assert provider.type == "image" + assert provider.images == ["alpine:3.18"] + assert provider.scanners == ["vuln", "secret"] + assert provider.image_config_scanners == [] + assert provider.trivy_severity == [] + assert provider.ignore_unfixed is False + assert provider.timeout == "5m" + assert provider.region == "container" + assert provider.audited_account == "image-scan" + assert provider.identity == "prowler" + assert provider.auth_method == "No auth" + assert provider.session is None + assert provider.audit_config == {} + assert provider.fixer_config == {} + assert provider._mutelist is None + + def test_image_provider_custom_params(self): + """Test initialization with custom parameters.""" + provider = _make_provider( + images=["nginx:1.25", "redis:7"], + scanners=["vuln", "secret", "misconfig"], + trivy_severity=["HIGH", "CRITICAL"], + ignore_unfixed=True, + timeout="10m", + fixer_config={"key": "value"}, + ) + + assert provider.images == ["nginx:1.25", "redis:7"] + assert provider.scanners == ["vuln", "secret", "misconfig"] + assert provider.trivy_severity == ["HIGH", "CRITICAL"] + assert provider.ignore_unfixed is True + assert provider.timeout == "10m" + assert provider.fixer_config == {"key": "value"} + + def test_image_provider_with_image_list_file(self): + """Test loading images from a file, skipping comments and blank lines.""" + with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f: + f.write("# Comment line\n") + f.write("alpine:3.18\n") + f.write("\n") + f.write(" nginx:latest \n") + f.write("# Another comment\n") + f.write("redis:7\n") + f.name + + provider = _make_provider( + images=None, + image_list_file=f.name, + ) + + assert "alpine:3.18" in provider.images + assert "nginx:latest" in provider.images + assert "redis:7" in provider.images + assert len(provider.images) == 3 + + def test_image_provider_no_images(self): + """Test that ImageNoImagesProvidedError is raised when no images are given.""" + with pytest.raises(ImageNoImagesProvidedError): + _make_provider(images=[]) + + def test_image_provider_image_list_file_not_found(self): + """Test that ImageListFileNotFoundError is raised for missing file.""" + with pytest.raises(ImageListFileNotFoundError): + _make_provider( + images=None, + image_list_file="/nonexistent/path/images.txt", + ) + + def test_process_finding_vulnerability(self): + """Test processing a vulnerability finding.""" + provider = _make_provider() + report = provider._process_finding( + SAMPLE_VULNERABILITY_FINDING, + "alpine:3.18", + "alpine:3.18 (alpine 3.18.0)", + image_sha="c1aabb73d233", + ) + + assert isinstance(report, CheckReportImage) + assert report.status == "FAIL" + assert report.check_metadata.CheckID == "CVE-2024-1234" + assert report.check_metadata.Severity == "high" + assert report.check_metadata.ServiceName == "container-image" + assert report.check_metadata.ResourceType == "container-image" + assert report.check_metadata.ResourceGroup == "container" + assert report.package_name == "openssl" + assert report.installed_version == "1.1.1k-r0" + assert report.fixed_version == "1.1.1l-r0" + assert report.resource_name == "alpine:3.18" + assert report.image_sha == "c1aabb73d233" + assert report.resource_details == "alpine:3.18 (alpine 3.18.0)" + assert report.region == "container" + assert report.check_metadata.Categories == ["vulnerability"] + assert report.check_metadata.RelatedUrl == "" + + def test_process_finding_secret(self): + """Test processing a secret finding (identified by RuleID).""" + provider = _make_provider() + report = provider._process_finding( + SAMPLE_SECRET_FINDING, + "myimage:latest", + "myimage:latest (debian 12)", + ) + + assert isinstance(report, CheckReportImage) + assert report.status == "FAIL" + assert report.check_metadata.CheckID == "aws-access-key-id" + assert report.check_metadata.Severity == "critical" + assert report.check_metadata.ServiceName == "container-image" + assert report.check_metadata.Categories == ["secrets"] + + def test_process_finding_misconfiguration(self): + """Test processing a misconfiguration finding (identified by ID).""" + provider = _make_provider() + report = provider._process_finding( + SAMPLE_MISCONFIGURATION_FINDING, + "myimage:latest", + "myimage:latest (debian 12)", + ) + + assert isinstance(report, CheckReportImage) + assert report.check_metadata.CheckID == "DS001" + assert report.check_metadata.Severity == "medium" + assert report.check_metadata.ServiceName == "container-image" + assert report.check_metadata.Categories == [] + + def test_process_finding_unknown_severity(self): + """Test that UNKNOWN severity is mapped to informational.""" + provider = _make_provider() + report = provider._process_finding( + SAMPLE_UNKNOWN_SEVERITY_FINDING, + "myimage:latest", + "myimage:latest (alpine 3.18.0)", + ) + + assert report.check_metadata.Severity == "informational" + + @patch("subprocess.run") + def test_run_scan_success(self, mock_subprocess): + """Test successful scan with mocked subprocess.""" + provider = _make_provider() + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_sample_trivy_json_output(), stderr="" + ) + + reports = [] + for batch in provider.run_scan(): + reports.extend(batch) + + assert len(reports) == 1 + assert reports[0].check_metadata.CheckID == "CVE-2024-1234" + assert reports[0].image_sha == SAMPLE_IMAGE_SHA + assert reports[0].resource_name == "alpine:3.18" + assert reports[0].check_metadata.ServiceName == "container-image" + + @patch("subprocess.run") + def test_run_scan_empty_output(self, mock_subprocess): + """Test scan with empty Trivy output produces no findings.""" + provider = _make_provider() + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_empty_trivy_output(), stderr="" + ) + + reports = [] + for batch in provider.run_scan(): + reports.extend(batch) + + assert len(reports) == 0 + + @patch("subprocess.run") + def test_run_scan_invalid_json(self, mock_subprocess): + """Test scan with malformed output doesn't crash.""" + provider = _make_provider() + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_invalid_trivy_output(), stderr="" + ) + + reports = [] + for batch in provider.run_scan(): + reports.extend(batch) + + assert len(reports) == 0 + + @patch("subprocess.run") + def test_run_scan_trivy_not_found(self, mock_subprocess): + """Test that ImageTrivyBinaryNotFoundError is raised when trivy is missing.""" + provider = _make_provider() + mock_subprocess.side_effect = FileNotFoundError( + "[Errno 2] No such file or directory: 'trivy'" + ) + + with pytest.raises(ImageTrivyBinaryNotFoundError): + for _ in provider._scan_single_image("alpine:3.18"): + pass + + @patch("subprocess.run") + def test_run_scan_multiple_images(self, mock_subprocess): + """Test scanning multiple images makes separate subprocess calls.""" + provider = _make_provider(images=["alpine:3.18", "nginx:latest"]) + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_sample_trivy_json_output(), stderr="" + ) + + reports = [] + for batch in provider.run_scan(): + reports.extend(batch) + + assert mock_subprocess.call_count == 2 + + @patch("subprocess.run") + def test_run_scan_multi_type_output(self, mock_subprocess): + """Test scan with vulnerabilities, secrets, and misconfigurations.""" + provider = _make_provider() + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_multi_type_trivy_output(), stderr="" + ) + + reports = [] + for batch in provider.run_scan(): + reports.extend(batch) + + assert len(reports) == 3 + + check_ids = [r.check_metadata.CheckID for r in reports] + assert "CVE-2024-1234" in check_ids + assert "aws-access-key-id" in check_ids + assert "DS001" in check_ids + + def test_print_credentials(self): + """Test that print_credentials outputs image names.""" + provider = _make_provider() + 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 "alpine:3.18" in output + + @patch("subprocess.run") + def test_test_connection_success(self, mock_subprocess): + """Test successful connection returns is_connected=True.""" + mock_subprocess.return_value = MagicMock(returncode=0, stderr="") + + result = ImageProvider.test_connection(image="alpine:3.18") + + assert result.is_connected is True + + @patch("subprocess.run") + def test_test_connection_auth_failure(self, mock_subprocess): + """Test 401 error returns auth failure.""" + mock_subprocess.return_value = MagicMock( + returncode=1, stderr="401 unauthorized" + ) + + result = ImageProvider.test_connection(image="private/image:latest") + + assert result.is_connected is False + assert "Authentication failed" in result.error + + @patch("subprocess.run") + def test_test_connection_not_found(self, mock_subprocess): + """Test 404 error returns not found.""" + mock_subprocess.return_value = MagicMock(returncode=1, stderr="404 not found") + + result = ImageProvider.test_connection(image="nonexistent/image:latest") + + assert result.is_connected is False + assert "not found" in result.error + + def test_build_status_extended(self): + """Test status message content for different finding types.""" + provider = _make_provider() + + # Vulnerability with fix + status = provider._build_status_extended(SAMPLE_VULNERABILITY_FINDING) + assert "CVE-2024-1234" in status + assert "openssl" in status + assert "fix available" in status + + # Finding with no special fields + status = provider._build_status_extended({"Description": "Simple finding"}) + assert status == "Simple finding" + + # Finding with will_not_fix status + finding_no_fix = { + "VulnerabilityID": "CVE-2024-0000", + "PkgName": "libc", + "Status": "will_not_fix", + "Title": "Some vuln", + } + status = provider._build_status_extended(finding_no_fix) + assert "no fix available" in status + + def test_validate_arguments(self): + """Test valid and invalid argument combinations.""" + # Valid: images provided + provider = _make_provider(images=["alpine:3.18"]) + assert provider.images == ["alpine:3.18"] + + # Invalid: empty images and no file + with pytest.raises(ImageNoImagesProvidedError): + _make_provider(images=[]) + + # Valid: custom scanners + provider = _make_provider(scanners=["vuln"]) + assert provider.scanners == ["vuln"] + + def test_setup_session(self): + """Test that setup_session returns None.""" + provider = _make_provider() + assert provider.setup_session() is None + + @patch("subprocess.run") + def test_run_method(self, mock_subprocess): + """Test that run() collects all batches into a list.""" + provider = _make_provider() + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_sample_trivy_json_output(), stderr="" + ) + + reports = provider.run() + + assert isinstance(reports, list) + assert len(reports) == 1 + + @patch("subprocess.run") + def test_scan_single_image_trivy_nonzero_exit(self, mock_subprocess): + """Test that a non-zero Trivy exit code raises ImageScanError.""" + provider = _make_provider() + mock_subprocess.return_value = MagicMock( + returncode=1, + stdout="", + stderr="fatal error: unable to pull image", + ) + + with pytest.raises(ImageScanError): + for _ in provider._scan_single_image("alpine:3.18"): + pass + + @patch("subprocess.run") + def test_scan_single_image_auth_failure(self, mock_subprocess): + """Test that a 401 unauthorized stderr raises ImageScanError with message.""" + provider = _make_provider() + mock_subprocess.return_value = MagicMock( + returncode=1, + stdout="", + stderr="ERROR 401 unauthorized: authentication required", + ) + + with pytest.raises(ImageScanError, match="401 unauthorized"): + for _ in provider._scan_single_image("private/image:latest"): + pass + + @patch("subprocess.run") + def test_sha_extraction_from_image_id(self, mock_subprocess): + """Test that image_sha is extracted from Trivy Metadata.ImageID.""" + provider = _make_provider() + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_sample_trivy_json_output(), stderr="" + ) + + reports = [] + for batch in provider._scan_single_image("alpine:3.18"): + reports.extend(batch) + + assert len(reports) == 1 + assert reports[0].image_sha == SAMPLE_IMAGE_SHA + + @patch("subprocess.run") + def test_sha_extraction_fallback_to_repo_digests(self, mock_subprocess): + """Test that image_sha falls back to RepoDigests when ImageID is absent.""" + provider = _make_provider() + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_repo_digest_only_trivy_output(), stderr="" + ) + + reports = [] + for batch in provider._scan_single_image("alpine:3.18"): + reports.extend(batch) + + assert len(reports) == 1 + assert reports[0].image_sha == "e5f6g7h8i9j0" + + @patch("subprocess.run") + def test_sha_extraction_no_metadata(self, mock_subprocess): + """Test that image_sha is empty when no Metadata is present.""" + provider = _make_provider() + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_no_metadata_trivy_output(), stderr="" + ) + + reports = [] + for batch in provider._scan_single_image("alpine:3.18"): + reports.extend(batch) + + assert len(reports) == 1 + assert reports[0].image_sha == "" + + @patch("subprocess.run") + def test_run_scan_propagates_scan_error(self, mock_subprocess): + """Test that run_scan() re-raises ImageScanError instead of swallowing it.""" + provider = _make_provider() + mock_subprocess.return_value = MagicMock( + returncode=1, + stdout="", + stderr="image not found", + ) + + with pytest.raises(ImageScanError): + for _ in provider.run_scan(): + pass + + +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 == "Docker login" + + 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 == "Docker login" + + @patch.dict( + os.environ, {"REGISTRY_USERNAME": "envuser", "REGISTRY_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 == "Docker login" + + @patch.dict(os.environ, {"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, {"REGISTRY_USERNAME": "envuser", "REGISTRY_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_sets_env_vars(self): + """Test that _build_trivy_env injects TRIVY_USERNAME/PASSWORD for native Trivy auth.""" + 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_sets_trivy_env_with_basic_auth(self, mock_subprocess): + """Test that _execute_trivy sets TRIVY_USERNAME/PASSWORD for native Trivy auth.""" + provider = _make_provider( + registry_username="myuser", + registry_password="mypass", + ) + mock_subprocess.return_value = MagicMock( + returncode=0, 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 uses Trivy native auth with TRIVY_USERNAME/PASSWORD env vars.""" + 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 + assert mock_subprocess.call_count == 1 + trivy_call = mock_subprocess.call_args + assert trivy_call.args[0][0] == "trivy" + env = trivy_call.kwargs.get("env") or trivy_call[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 "Docker login" in output + + +class TestExtractRegistry: + def test_docker_hub_simple(self): + assert ImageProvider._extract_registry("alpine:3.18") is None + + def test_docker_hub_with_namespace(self): + assert ImageProvider._extract_registry("andoniaf/test-private:tag") is None + + def test_ghcr(self): + assert ImageProvider._extract_registry("ghcr.io/user/image:tag") == "ghcr.io" + + def test_ecr(self): + assert ( + ImageProvider._extract_registry( + "123456789012.dkr.ecr.us-east-1.amazonaws.com/repo:tag" + ) + == "123456789012.dkr.ecr.us-east-1.amazonaws.com" + ) + + def test_localhost_with_port(self): + assert ( + ImageProvider._extract_registry("localhost:5000/myimage:latest") + == "localhost:5000" + ) + + def test_custom_registry_with_port(self): + assert ( + ImageProvider._extract_registry("myregistry.io:5000/image:tag") + == "myregistry.io:5000" + ) + + def test_digest_reference(self): + assert ( + ImageProvider._extract_registry("ghcr.io/user/image@sha256:abc123") + == "ghcr.io" + ) + + def test_bare_image_name(self): + assert ImageProvider._extract_registry("nginx") is None + + +class TestCleanup: + def test_cleanup_idempotent(self): + """Test cleanup is safe to call multiple times.""" + provider = _make_provider() + + provider.cleanup() + provider.cleanup() + + +class TestImageProviderInputValidation: + def test_invalid_timeout_format_raises_error(self): + """Test that a non-matching timeout string raises ImageInvalidTimeoutError.""" + with pytest.raises(ImageInvalidTimeoutError): + _make_provider(timeout="invalid") + + def test_invalid_timeout_no_unit_raises_error(self): + """Test that a numeric timeout without a unit raises ImageInvalidTimeoutError.""" + with pytest.raises(ImageInvalidTimeoutError): + _make_provider(timeout="300") + + def test_invalid_timeout_wrong_unit_raises_error(self): + """Test that a timeout with an unsupported unit raises ImageInvalidTimeoutError.""" + with pytest.raises(ImageInvalidTimeoutError): + _make_provider(timeout="5d") + + def test_valid_timeout_seconds(self): + """Test that a seconds-based timeout is accepted.""" + provider = _make_provider(timeout="300s") + assert provider.timeout == "300s" + + def test_valid_timeout_hours(self): + """Test that an hours-based timeout is accepted.""" + provider = _make_provider(timeout="1h") + assert provider.timeout == "1h" + + def test_invalid_scanner_raises_error(self): + """Test that an invalid scanner name raises ImageInvalidScannerError.""" + with pytest.raises(ImageInvalidScannerError): + _make_provider(scanners=["vuln", "bad"]) + + def test_invalid_severity_raises_error(self): + """Test that an invalid severity level raises ImageInvalidSeverityError.""" + with pytest.raises(ImageInvalidSeverityError): + _make_provider(trivy_severity=["HIGH", "SUPER_HIGH"]) + + def test_valid_all_scanners(self): + """Test that all valid scanner choices are accepted.""" + provider = _make_provider(scanners=["vuln", "secret", "misconfig", "license"]) + assert provider.scanners == ["vuln", "secret", "misconfig", "license"] + + def test_valid_all_severities(self): + """Test that all valid severity choices are accepted.""" + provider = _make_provider( + trivy_severity=["CRITICAL", "HIGH", "MEDIUM", "LOW", "UNKNOWN"] + ) + assert provider.trivy_severity == [ + "CRITICAL", + "HIGH", + "MEDIUM", + "LOW", + "UNKNOWN", + ] + + def test_image_config_scanners_defaults_to_empty(self): + """Test that image_config_scanners defaults to an empty list.""" + provider = _make_provider() + assert provider.image_config_scanners == [] + + def test_valid_image_config_scanners(self): + """Test that valid image config scanners are accepted.""" + provider = _make_provider(image_config_scanners=["misconfig", "secret"]) + assert provider.image_config_scanners == ["misconfig", "secret"] + + def test_invalid_image_config_scanner_raises_error(self): + """Test that an invalid image config scanner raises ImageInvalidConfigScannerError.""" + with pytest.raises(ImageInvalidConfigScannerError): + _make_provider(image_config_scanners=["misconfig", "vuln"]) + + @patch("subprocess.run") + def test_trivy_command_includes_image_config_scanners(self, mock_subprocess): + """Test that Trivy command includes --image-config-scanners when set.""" + provider = _make_provider(image_config_scanners=["misconfig", "secret"]) + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_empty_trivy_output(), stderr="" + ) + + for _ in provider._scan_single_image("alpine:3.18"): + pass + + call_args = mock_subprocess.call_args[0][0] + assert "--image-config-scanners" in call_args + idx = call_args.index("--image-config-scanners") + assert call_args[idx + 1] == "misconfig,secret" + + @patch("subprocess.run") + def test_trivy_command_omits_image_config_scanners_when_empty( + self, mock_subprocess + ): + """Test that Trivy command omits --image-config-scanners when empty.""" + provider = _make_provider(image_config_scanners=[]) + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_empty_trivy_output(), stderr="" + ) + + for _ in provider._scan_single_image("alpine:3.18"): + pass + + call_args = mock_subprocess.call_args[0][0] + assert "--image-config-scanners" not in call_args + + +class TestImageProviderErrorCategorization: + def test_categorize_auth_failure(self): + """Test that auth-related errors are categorized correctly.""" + result = ImageProvider._categorize_trivy_error( + "401 unauthorized: access denied" + ) + assert "Auth failure" in result + + def test_categorize_not_found(self): + """Test that not-found errors are categorized correctly.""" + result = ImageProvider._categorize_trivy_error( + "manifest unknown: image not found" + ) + assert "Image not found" in result + + def test_categorize_rate_limit(self): + """Test that rate-limit errors are categorized correctly.""" + result = ImageProvider._categorize_trivy_error("429 too many requests") + assert "Rate limited" in result + + def test_categorize_network_issue(self): + """Test that network errors are categorized correctly.""" + result = ImageProvider._categorize_trivy_error("connection refused to registry") + assert "Network issue" in result + + def test_categorize_unknown_error(self): + """Test that unrecognized errors are returned as-is.""" + msg = "some unknown trivy error" + result = ImageProvider._categorize_trivy_error(msg) + assert result == msg + + +class TestImageProviderNameValidation: + @pytest.mark.parametrize( + "bad_name", + [ + "alpine;rm -rf /", + "image|cat /etc/passwd", + "image&background", + "image$VAR", + "image`whoami`", + "image\ninjected", + "image\rinjected", + ], + ) + def test_image_provider_invalid_image_name_shell_chars(self, bad_name): + """Test that image names with shell metacharacters raise ImageInvalidNameError.""" + with pytest.raises(ImageInvalidNameError): + _make_provider(images=[bad_name]) + + def test_image_provider_invalid_image_name_empty(self): + """Test that an empty string image name raises ImageInvalidNameError.""" + with pytest.raises(ImageInvalidNameError): + _make_provider(images=[""]) + + @pytest.mark.parametrize( + "valid_name", + [ + "alpine:3.18", + "nginx:latest", + "registry.example.com/repo/image:tag", + "ghcr.io/owner/image:v1.2.3", + "myimage@sha256:abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890", + "localhost:5000/myimage:latest", + ], + ) + def test_image_provider_valid_image_names(self, valid_name): + """Test that various valid image name formats pass validation.""" + provider = _make_provider(images=[valid_name]) + assert valid_name in provider.images + + def test_image_provider_image_name_too_long(self): + """Test that a name exceeding 500 chars raises ImageInvalidNameError.""" + long_name = "a" * 501 + with pytest.raises(ImageInvalidNameError): + _make_provider(images=[long_name]) + + def test_image_provider_file_too_many_lines(self): + """Test that a file with more than MAX_IMAGE_LIST_LINES raises ImageListFileReadError.""" + with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f: + for i in range(10_001): + f.write(f"image{i}:latest\n") + f.flush() + file_path = f.name + + with pytest.raises(ImageListFileReadError): + _make_provider(images=None, image_list_file=file_path) + + +class TestScanPerImage: + @patch("subprocess.run") + def test_yields_per_image(self, mock_subprocess): + """Test that scan_per_image yields (name, findings) per image.""" + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_sample_trivy_json_output(), stderr="" + ) + provider = _make_provider(images=["alpine:3.18", "nginx:latest"]) + + results = list(provider.scan_per_image()) + + assert len(results) == 2 + for name, findings in results: + assert isinstance(name, str) + assert isinstance(findings, list) + assert all(isinstance(f, CheckReportImage) for f in findings) + + @patch("subprocess.run") + def test_reraises_scan_error(self, mock_subprocess): + """Test that ImageScanError propagates from scan_per_image.""" + mock_subprocess.return_value = MagicMock( + returncode=1, stdout="", stderr="scan failed" + ) + provider = _make_provider(images=["alpine:3.18"]) + + with pytest.raises(ImageScanError): + list(provider.scan_per_image()) + + @patch("subprocess.run") + def test_skips_generic_error(self, mock_subprocess): + """Test that a generic RuntimeError in _scan_single_image yields empty findings and continues.""" + + def side_effect(cmd, **kwargs): + if "bad:image" in cmd: + raise RuntimeError("unexpected error") + return MagicMock( + returncode=0, stdout=get_sample_trivy_json_output(), stderr="" + ) + + mock_subprocess.side_effect = side_effect + provider = _make_provider(images=["bad:image", "alpine:3.18"]) + + results = list(provider.scan_per_image()) + + assert len(results) == 2 + assert results[0][0] == "bad:image" + assert results[0][1] == [] + assert results[1][0] == "alpine:3.18" + assert len(results[1][1]) > 0 + + @patch("subprocess.run") + def test_calls_cleanup(self, mock_subprocess): + """Test that cleanup is called even after scan_per_image completes.""" + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_sample_trivy_json_output(), stderr="" + ) + provider = _make_provider(images=["alpine:3.18"]) + + with mock.patch.object(provider, "cleanup") as mock_cleanup: + list(provider.scan_per_image()) + + mock_cleanup.assert_called_once() diff --git a/tests/providers/image/lib/__init__.py b/tests/providers/image/lib/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/providers/image/lib/registry/__init__.py b/tests/providers/image/lib/registry/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/providers/image/lib/registry/test_arguments.py b/tests/providers/image/lib/registry/test_arguments.py new file mode 100644 index 0000000000..5b52d385fd --- /dev/null +++ b/tests/providers/image/lib/registry/test_arguments.py @@ -0,0 +1,223 @@ +from argparse import Namespace + +from prowler.providers.image.lib.arguments.arguments import validate_arguments + + +class TestValidateArguments: + def test_no_source_fails(self): + args = Namespace( + images=[], + image_list_file=None, + registry=None, + image_filter=None, + tag_filter=None, + max_images=0, + registry_insecure=False, + registry_list_images=False, + ) + ok, msg = validate_arguments(args) + assert not ok + assert "--image" in msg + + def test_image_only_passes(self): + args = Namespace( + images=["nginx:latest"], + image_list_file=None, + registry=None, + image_filter=None, + tag_filter=None, + max_images=0, + registry_insecure=False, + registry_list_images=False, + ) + ok, _ = validate_arguments(args) + assert ok + + def test_image_list_only_passes(self): + args = Namespace( + images=[], + image_list_file="images.txt", + registry=None, + image_filter=None, + tag_filter=None, + max_images=0, + registry_insecure=False, + registry_list_images=False, + ) + ok, _ = validate_arguments(args) + assert ok + + def test_registry_only_passes(self): + args = Namespace( + images=[], + image_list_file=None, + registry="myregistry.io", + image_filter=None, + tag_filter=None, + max_images=0, + registry_insecure=False, + registry_list_images=False, + ) + ok, _ = validate_arguments(args) + assert ok + + def test_image_filter_without_registry_fails(self): + args = Namespace( + images=["nginx:latest"], + image_list_file=None, + registry=None, + image_filter="^prod", + tag_filter=None, + max_images=0, + registry_insecure=False, + registry_list_images=False, + ) + ok, msg = validate_arguments(args) + assert not ok + assert "--image-filter requires --registry" in msg + + def test_tag_filter_without_registry_fails(self): + args = Namespace( + images=["nginx:latest"], + image_list_file=None, + registry=None, + image_filter=None, + tag_filter="^v", + max_images=0, + registry_insecure=False, + registry_list_images=False, + ) + ok, msg = validate_arguments(args) + assert not ok + assert "--tag-filter requires --registry" in msg + + def test_max_images_without_registry_fails(self): + args = Namespace( + images=["nginx:latest"], + image_list_file=None, + registry=None, + image_filter=None, + tag_filter=None, + max_images=50, + registry_insecure=False, + registry_list_images=False, + ) + ok, msg = validate_arguments(args) + assert not ok + assert "--max-images requires --registry" in msg + + def test_registry_insecure_without_registry_fails(self): + args = Namespace( + images=[], + image_list_file="i.txt", + registry=None, + image_filter=None, + tag_filter=None, + max_images=0, + registry_insecure=True, + registry_list_images=False, + ) + ok, msg = validate_arguments(args) + assert not ok + assert "--registry-insecure requires --registry" in msg + + def test_docker_hub_no_namespace_fails(self): + args = Namespace( + images=[], + image_list_file=None, + registry="docker.io", + image_filter=None, + tag_filter=None, + max_images=0, + registry_insecure=False, + registry_list_images=False, + ) + ok, msg = validate_arguments(args) + assert not ok + assert "namespace" in msg.lower() + + def test_docker_hub_with_namespace_passes(self): + args = Namespace( + images=[], + image_list_file=None, + registry="docker.io/myorg", + image_filter=None, + tag_filter=None, + max_images=0, + registry_insecure=False, + registry_list_images=False, + ) + ok, _ = validate_arguments(args) + assert ok + + def test_docker_hub_https_no_namespace_fails(self): + args = Namespace( + images=[], + image_list_file=None, + registry="https://docker.io", + image_filter=None, + tag_filter=None, + max_images=0, + registry_insecure=False, + registry_list_images=False, + ) + ok, msg = validate_arguments(args) + assert not ok + assert "namespace" in msg.lower() + + def test_registry_with_filters_passes(self): + args = Namespace( + images=[], + image_list_file=None, + registry="myregistry.io", + image_filter="^prod", + tag_filter="^v", + max_images=100, + registry_insecure=True, + registry_list_images=False, + ) + ok, _ = validate_arguments(args) + assert ok + + def test_registry_list_without_registry_fails(self): + args = Namespace( + images=["nginx:latest"], + image_list_file=None, + registry=None, + image_filter=None, + tag_filter=None, + max_images=0, + registry_insecure=False, + registry_list_images=True, + ) + ok, msg = validate_arguments(args) + assert not ok + assert "--registry-list requires --registry" in msg + + def test_registry_list_with_registry_passes(self): + args = Namespace( + images=[], + image_list_file=None, + registry="myregistry.io", + image_filter=None, + tag_filter=None, + max_images=0, + registry_insecure=False, + registry_list_images=True, + ) + ok, _ = validate_arguments(args) + assert ok + + def test_combined_registry_and_image_passes(self): + args = Namespace( + images=["nginx:latest"], + image_list_file=None, + registry="myregistry.io", + image_filter=None, + tag_filter=None, + max_images=0, + registry_insecure=False, + registry_list_images=False, + ) + ok, _ = validate_arguments(args) + assert ok diff --git a/tests/providers/image/lib/registry/test_dockerhub_adapter.py b/tests/providers/image/lib/registry/test_dockerhub_adapter.py new file mode 100644 index 0000000000..930e872bba --- /dev/null +++ b/tests/providers/image/lib/registry/test_dockerhub_adapter.py @@ -0,0 +1,243 @@ +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from prowler.providers.image.exceptions.exceptions import ( + ImageRegistryAuthError, + ImageRegistryCatalogError, + ImageRegistryNetworkError, +) +from prowler.providers.image.lib.registry.dockerhub_adapter import DockerHubAdapter + + +class TestDockerHubAdapterInit: + def test_extract_namespace_simple(self): + assert DockerHubAdapter._extract_namespace("docker.io/myorg") == "myorg" + + def test_extract_namespace_https(self): + assert DockerHubAdapter._extract_namespace("https://docker.io/myorg") == "myorg" + + def test_extract_namespace_registry1(self): + assert ( + DockerHubAdapter._extract_namespace("registry-1.docker.io/myorg") == "myorg" + ) + + def test_extract_namespace_empty(self): + assert DockerHubAdapter._extract_namespace("docker.io") == "" + + def test_extract_namespace_with_slash(self): + assert DockerHubAdapter._extract_namespace("docker.io/myorg/") == "myorg" + + +class TestDockerHubListRepositories: + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_list_repos(self, mock_request): + # Hub login (now goes through requests.request via _request_with_retry) + login_resp = MagicMock(status_code=200) + login_resp.json.return_value = {"token": "jwt"} + # Repo listing + repos_resp = MagicMock(status_code=200) + repos_resp.json.return_value = { + "results": [{"name": "app1"}, {"name": "app2"}], + "next": None, + } + mock_request.side_effect = [login_resp, repos_resp] + adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p") + repos = adapter.list_repositories() + assert repos == ["myorg/app1", "myorg/app2"] + + def test_list_repos_no_namespace_raises(self): + adapter = DockerHubAdapter("docker.io") + with pytest.raises(ImageRegistryCatalogError, match="namespace"): + adapter.list_repositories() + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_list_repos_public_no_credentials(self, mock_request): + """When no credentials are provided, use the public /v2/repositories/{ns}/ endpoint.""" + repos_resp = MagicMock(status_code=200) + repos_resp.json.return_value = { + "results": [{"name": "repo1"}, {"name": "repo2"}], + "next": None, + } + mock_request.return_value = repos_resp + adapter = DockerHubAdapter("docker.io/publicns") + repos = adapter.list_repositories() + assert repos == ["publicns/repo1", "publicns/repo2"] + called_url = mock_request.call_args[0][1] + assert "/v2/repositories/publicns/" in called_url + assert "/v2/namespaces/" not in called_url + + +class TestDockerHubListTags: + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_list_tags(self, mock_request): + # Token exchange (now goes through requests.request via _request_with_retry) + token_resp = MagicMock(status_code=200) + token_resp.json.return_value = {"token": "registry-token"} + # Tag listing + tags_resp = MagicMock(status_code=200, headers={}) + tags_resp.json.return_value = {"tags": ["latest", "v1.0"]} + mock_request.side_effect = [token_resp, tags_resp] + adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p") + tags = adapter.list_tags("myorg/myapp") + assert tags == ["latest", "v1.0"] + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_list_tags_auth_failure(self, mock_request): + # Token exchange + token_resp = MagicMock(status_code=200) + token_resp.json.return_value = {"token": "tok"} + # Tag listing returns 401 + tags_resp = MagicMock(status_code=401) + mock_request.side_effect = [token_resp, tags_resp] + adapter = DockerHubAdapter("docker.io/myorg") + with pytest.raises(ImageRegistryAuthError): + adapter.list_tags("myorg/myapp") + + +class TestDockerHubLogin: + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_login_failure(self, mock_request): + resp = MagicMock(status_code=401, text="invalid credentials") + mock_request.return_value = resp + adapter = DockerHubAdapter("docker.io/myorg", username="bad", password="creds") + with pytest.raises(ImageRegistryAuthError, match="login failed"): + adapter._hub_login() + + def test_login_skipped_without_credentials(self): + adapter = DockerHubAdapter("docker.io/myorg") + adapter._hub_login() # Should not raise + assert adapter._hub_jwt is None + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_login_401_includes_response_body(self, mock_request): + resp = MagicMock( + status_code=401, text='{"detail":"Incorrect authentication credentials"}' + ) + mock_request.return_value = resp + adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p") + with pytest.raises( + ImageRegistryAuthError, match="Incorrect authentication credentials" + ): + adapter._hub_login() + + @patch("prowler.providers.image.lib.registry.base.time.sleep") + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_login_500_retried_then_raises_network_error( + self, mock_request, mock_sleep + ): + mock_request.return_value = MagicMock(status_code=500) + adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p") + with pytest.raises(ImageRegistryNetworkError, match="Server error"): + adapter._hub_login() + assert mock_request.call_count == 3 + + +class TestDockerHubRetry: + @patch("prowler.providers.image.lib.registry.base.time.sleep") + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_retry_on_429(self, mock_request, mock_sleep): + resp_429 = MagicMock(status_code=429) + resp_200 = MagicMock(status_code=200) + mock_request.side_effect = [resp_429, resp_200] + adapter = DockerHubAdapter("docker.io/myorg") + result = adapter._request_with_retry( + "GET", "https://hub.docker.com/v2/namespaces/myorg/repositories" + ) + assert result.status_code == 200 + + @patch("prowler.providers.image.lib.registry.base.time.sleep") + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_connection_error_retries(self, mock_request, mock_sleep): + mock_request.side_effect = requests.exceptions.ConnectionError("fail") + adapter = DockerHubAdapter("docker.io/myorg") + with pytest.raises(ImageRegistryNetworkError): + adapter._request_with_retry("GET", "https://hub.docker.com") + assert mock_request.call_count == 3 + + @patch("prowler.providers.image.lib.registry.base.time.sleep") + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_retry_on_500(self, mock_request, mock_sleep): + resp_500 = MagicMock(status_code=500) + resp_200 = MagicMock(status_code=200) + mock_request.side_effect = [resp_500, resp_200] + adapter = DockerHubAdapter("docker.io/myorg") + result = adapter._request_with_retry("GET", "https://hub.docker.com") + assert result.status_code == 200 + assert mock_request.call_count == 2 + mock_sleep.assert_called_once() + + @patch("prowler.providers.image.lib.registry.base.time.sleep") + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_retry_exhausted_on_500_raises_network_error( + self, mock_request, mock_sleep + ): + mock_request.return_value = MagicMock(status_code=500) + adapter = DockerHubAdapter("docker.io/myorg") + with pytest.raises( + ImageRegistryNetworkError, match="Server error.*HTTP 500.*3 attempts" + ): + adapter._request_with_retry("GET", "https://hub.docker.com") + assert mock_request.call_count == 3 + + @patch("prowler.providers.image.lib.registry.base.time.sleep") + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_4xx_not_retried(self, mock_request, mock_sleep): + mock_request.return_value = MagicMock(status_code=403) + adapter = DockerHubAdapter("docker.io/myorg") + result = adapter._request_with_retry("GET", "https://hub.docker.com") + assert result.status_code == 403 + assert mock_request.call_count == 1 + mock_sleep.assert_not_called() + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_request_sends_user_agent(self, mock_request): + mock_request.return_value = MagicMock(status_code=200) + adapter = DockerHubAdapter("docker.io/myorg") + adapter._request_with_retry("GET", "https://hub.docker.com") + _, kwargs = mock_request.call_args + from prowler.config.config import prowler_version + + assert ( + kwargs["headers"]["User-Agent"] + == f"Prowler/{prowler_version} (registry-adapter)" + ) + + @patch("prowler.providers.image.lib.registry.base.time.sleep") + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_retry_500_includes_response_body(self, mock_request, mock_sleep): + resp_500 = MagicMock(status_code=500, text="Cloudflare error") + mock_request.return_value = resp_500 + adapter = DockerHubAdapter("docker.io/myorg") + with pytest.raises(ImageRegistryNetworkError, match="Cloudflare error"): + adapter._request_with_retry("GET", "https://hub.docker.com") + + +class TestDockerHubEmptyTokens: + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_empty_hub_jwt_raises(self, mock_request): + resp = MagicMock(status_code=200) + resp.json.return_value = {"token": ""} + mock_request.return_value = resp + adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p") + with pytest.raises(ImageRegistryAuthError, match="empty JWT"): + adapter._hub_login() + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_none_hub_jwt_raises(self, mock_request): + resp = MagicMock(status_code=200) + resp.json.return_value = {} + mock_request.return_value = resp + adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p") + with pytest.raises(ImageRegistryAuthError, match="empty JWT"): + adapter._hub_login() + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_empty_registry_token_raises(self, mock_request): + resp = MagicMock(status_code=200) + resp.json.return_value = {"token": ""} + mock_request.return_value = resp + adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p") + with pytest.raises(ImageRegistryAuthError, match="empty token"): + adapter._get_registry_token("myorg/myapp") diff --git a/tests/providers/image/lib/registry/test_factory.py b/tests/providers/image/lib/registry/test_factory.py new file mode 100644 index 0000000000..74aab82d02 --- /dev/null +++ b/tests/providers/image/lib/registry/test_factory.py @@ -0,0 +1,34 @@ +from prowler.providers.image.lib.registry.dockerhub_adapter import DockerHubAdapter +from prowler.providers.image.lib.registry.factory import create_registry_adapter +from prowler.providers.image.lib.registry.oci_adapter import OciRegistryAdapter + + +class TestCreateRegistryAdapter: + def test_docker_hub_returns_dockerhub_adapter(self): + adapter = create_registry_adapter("docker.io/myorg") + assert isinstance(adapter, DockerHubAdapter) + + def test_oci_returns_oci_adapter(self): + adapter = create_registry_adapter("myregistry.io") + assert isinstance(adapter, OciRegistryAdapter) + + def test_ecr_returns_oci_adapter(self): + adapter = create_registry_adapter("123456789.dkr.ecr.us-east-1.amazonaws.com") + assert isinstance(adapter, OciRegistryAdapter) + + def test_passes_credentials(self): + adapter = create_registry_adapter( + "myregistry.io", + username="user", + password="pass", + token="tok", + verify_ssl=False, + ) + assert adapter.username == "user" + assert adapter.password == "pass" + assert adapter.token == "tok" + assert adapter.verify_ssl is False + + def test_registry_1_docker_io(self): + adapter = create_registry_adapter("registry-1.docker.io/myorg") + assert isinstance(adapter, DockerHubAdapter) diff --git a/tests/providers/image/lib/registry/test_oci_adapter.py b/tests/providers/image/lib/registry/test_oci_adapter.py new file mode 100644 index 0000000000..32b0202e2d --- /dev/null +++ b/tests/providers/image/lib/registry/test_oci_adapter.py @@ -0,0 +1,418 @@ +import base64 +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from prowler.providers.image.exceptions.exceptions import ( + ImageRegistryAuthError, + ImageRegistryCatalogError, + ImageRegistryNetworkError, +) +from prowler.providers.image.lib.registry.oci_adapter import OciRegistryAdapter + + +class TestOciAdapterInit: + def test_normalise_url_adds_https(self): + adapter = OciRegistryAdapter("myregistry.io") + assert adapter._base_url == "https://myregistry.io" + + def test_normalise_url_keeps_http(self): + adapter = OciRegistryAdapter("http://myregistry.io") + assert adapter._base_url == "http://myregistry.io" + + def test_normalise_url_strips_trailing_slash(self): + adapter = OciRegistryAdapter("https://myregistry.io/") + 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 + ) + assert adapter.username == "u" + assert adapter.password == "p" + assert adapter.token == "t" + assert adapter.verify_ssl is False + + +class TestOciAdapterAuth: + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_ensure_auth_with_token(self, mock_request): + adapter = OciRegistryAdapter("reg.io", token="my-token") + adapter._ensure_auth() + assert adapter._bearer_token == "my-token" + mock_request.assert_not_called() + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_ensure_auth_anonymous_ok(self, mock_request): + resp = MagicMock(status_code=200) + mock_request.return_value = resp + adapter = OciRegistryAdapter("reg.io") + adapter._ensure_auth() + assert adapter._bearer_token is None + + @patch("prowler.providers.image.lib.registry.base.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"' + }, + ) + token_resp = MagicMock(status_code=200) + token_resp.json.return_value = {"token": "bearer-tok"} + mock_request.side_effect = [ping_resp, token_resp] + adapter = OciRegistryAdapter("reg.io", username="u", password="p") + adapter._ensure_auth() + assert adapter._bearer_token == "bearer-tok" + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_ensure_auth_403_raises(self, mock_request): + resp = MagicMock(status_code=403) + mock_request.return_value = resp + adapter = OciRegistryAdapter("reg.io") + with pytest.raises(ImageRegistryAuthError): + adapter._ensure_auth() + + @patch("prowler.providers.image.lib.registry.base.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.base.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.base.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.base.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) + + 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") + 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"] + } + mock_request.side_effect = [ping_resp, catalog_resp] + adapter = OciRegistryAdapter("reg.io") + repos = adapter.list_repositories() + assert repos == ["app/frontend", "app/backend"] + + @patch("prowler.providers.image.lib.registry.base.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.json.return_value = {"repositories": ["a"]} + page2_resp = MagicMock(status_code=200, headers={}) + page2_resp.json.return_value = {"repositories": ["b"]} + mock_request.side_effect = [ping_resp, page1_resp, page2_resp] + adapter = OciRegistryAdapter("reg.io") + repos = adapter.list_repositories() + assert repos == ["a", "b"] + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_list_repos_404_raises(self, mock_request): + ping_resp = MagicMock(status_code=200) + catalog_resp = MagicMock(status_code=404) + mock_request.side_effect = [ping_resp, catalog_resp] + adapter = OciRegistryAdapter("reg.io") + with pytest.raises(ImageRegistryCatalogError): + adapter.list_repositories() + + +class TestOciAdapterListTags: + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_list_tags(self, mock_request): + ping_resp = MagicMock(status_code=200) + tags_resp = MagicMock(status_code=200, headers={}) + tags_resp.json.return_value = {"tags": ["latest", "v1.0"]} + mock_request.side_effect = [ping_resp, tags_resp] + adapter = OciRegistryAdapter("reg.io") + tags = adapter.list_tags("myapp") + assert tags == ["latest", "v1.0"] + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_list_tags_null_tags(self, mock_request): + ping_resp = MagicMock(status_code=200) + tags_resp = MagicMock(status_code=200, headers={}) + tags_resp.json.return_value = {"tags": None} + mock_request.side_effect = [ping_resp, tags_resp] + adapter = OciRegistryAdapter("reg.io") + tags = adapter.list_tags("myapp") + assert tags == [] + + +class TestOciAdapterRetry: + @patch("prowler.providers.image.lib.registry.base.time.sleep") + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_retry_on_429(self, mock_request, mock_sleep): + resp_429 = MagicMock(status_code=429) + resp_200 = MagicMock(status_code=200) + mock_request.side_effect = [resp_429, resp_200] + adapter = OciRegistryAdapter("reg.io") + result = adapter._request_with_retry("GET", "https://reg.io/v2/") + assert result.status_code == 200 + mock_sleep.assert_called_once() + + @patch("prowler.providers.image.lib.registry.base.time.sleep") + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_connection_error_retries(self, mock_request, mock_sleep): + mock_request.side_effect = requests.exceptions.ConnectionError("failed") + adapter = OciRegistryAdapter("reg.io") + with pytest.raises(ImageRegistryNetworkError): + adapter._request_with_retry("GET", "https://reg.io/v2/") + assert mock_request.call_count == 3 + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_timeout_raises_immediately(self, mock_request): + mock_request.side_effect = requests.exceptions.Timeout("timeout") + adapter = OciRegistryAdapter("reg.io") + with pytest.raises(ImageRegistryNetworkError): + adapter._request_with_retry("GET", "https://reg.io/v2/") + assert mock_request.call_count == 1 + + +class TestOciAdapterNextPageUrl: + def test_no_link_header(self): + resp = MagicMock(headers={}) + 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" + ) + + def test_link_header_no_next(self): + resp = MagicMock( + headers={"Link": '; rel="prev"'} + ) + assert OciRegistryAdapter._next_page_url(resp) is None + + +class TestOciAdapterSSRF: + def test_reject_file_scheme(self): + adapter = OciRegistryAdapter("reg.io") + with pytest.raises(ImageRegistryAuthError, match="disallowed scheme"): + adapter._validate_realm_url("file:///etc/passwd") + + def test_reject_ftp_scheme(self): + adapter = OciRegistryAdapter("reg.io") + with pytest.raises(ImageRegistryAuthError, match="disallowed scheme"): + adapter._validate_realm_url("ftp://evil.com/token") + + def test_reject_private_ip(self): + adapter = OciRegistryAdapter("reg.io") + with pytest.raises(ImageRegistryAuthError, match="private/loopback"): + adapter._validate_realm_url("https://10.0.0.1/token") + + def test_reject_loopback(self): + adapter = OciRegistryAdapter("reg.io") + with pytest.raises(ImageRegistryAuthError, match="private/loopback"): + adapter._validate_realm_url("https://127.0.0.1/token") + + def test_reject_link_local(self): + adapter = OciRegistryAdapter("reg.io") + with pytest.raises(ImageRegistryAuthError, match="private/loopback"): + adapter._validate_realm_url("https://169.254.169.254/latest/meta-data") + + def test_accept_public_https(self): + adapter = OciRegistryAdapter("reg.io") + # Should not raise + adapter._validate_realm_url("https://auth.example.com/token") + + def test_accept_hostname_not_ip(self): + adapter = OciRegistryAdapter("reg.io") + # Hostnames (not IPs) should pass even if they resolve to private IPs + adapter._validate_realm_url("https://internal.corp.com/token") + + +class TestOciAdapterEmptyToken: + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_empty_bearer_token_raises(self, mock_request): + 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": "", "access_token": ""} + mock_request.side_effect = [ping_resp, token_resp] + adapter = OciRegistryAdapter("reg.io", username="u", password="p") + with pytest.raises(ImageRegistryAuthError, match="empty token"): + adapter._ensure_auth() + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_none_bearer_token_raises(self, mock_request): + 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 = {} + mock_request.side_effect = [ping_resp, token_resp] + adapter = OciRegistryAdapter("reg.io", username="u", password="p") + with pytest.raises(ImageRegistryAuthError, match="empty token"): + adapter._ensure_auth() + + +class TestOciAdapterNarrowExcept: + def test_invalid_utf8_base64_falls_through(self): + # Create a base64 string that decodes to invalid UTF-8 + invalid_bytes = base64.b64encode(b"\xff\xfe").decode() + adapter = OciRegistryAdapter("ecr.aws", username="AWS", password=invalid_bytes) + user, pwd = adapter._resolve_basic_credentials() + assert user == "AWS" + assert pwd == invalid_bytes + + +class TestCredentialRedaction: + def test_getstate_redacts_credentials(self): + adapter = OciRegistryAdapter( + "reg.io", username="u", password="secret", token="tok" + ) + state = adapter.__getstate__() + assert state["_password"] == "***" + assert state["_token"] == "***" + assert state["username"] == "u" + assert state["registry_url"] == "reg.io" + + def test_getstate_none_credentials(self): + adapter = OciRegistryAdapter("reg.io") + state = adapter.__getstate__() + assert state["_password"] is None + assert state["_token"] is None + + def test_repr_redacts_credentials(self): + adapter = OciRegistryAdapter( + "reg.io", username="u", password="s3cret_pw", token="s3cret_tk" + ) + r = repr(adapter) + assert "s3cret_pw" not in r + assert "s3cret_tk" not in r + assert "" in r + + def test_properties_still_work(self): + adapter = OciRegistryAdapter("reg.io", password="secret", token="tok") + assert adapter.password == "secret" + assert adapter.token == "tok" diff --git a/tests/providers/image/lib/registry/test_provider_registry.py b/tests/providers/image/lib/registry/test_provider_registry.py new file mode 100644 index 0000000000..97901ea0c4 --- /dev/null +++ b/tests/providers/image/lib/registry/test_provider_registry.py @@ -0,0 +1,234 @@ +import os +from unittest.mock import MagicMock, patch + +import pytest + +from prowler.providers.image.exceptions.exceptions import ( + ImageInvalidFilterError, + ImageMaxImagesExceededError, +) +from prowler.providers.image.image_provider import ImageProvider +from prowler.providers.image.lib.registry.dockerhub_adapter import DockerHubAdapter + +_CLEAN_ENV = { + "PATH": os.environ.get("PATH", ""), + "HOME": os.environ.get("HOME", ""), +} + + +def _build_provider(**overrides): + defaults = dict( + images=[], + registry="myregistry.io", + image_filter=None, + tag_filter=None, + max_images=0, + registry_insecure=False, + registry_list_images=False, + config_content={"image": {}}, + ) + defaults.update(overrides) + with patch.dict(os.environ, _CLEAN_ENV, clear=True): + return ImageProvider(**defaults) + + +class TestRegistryEnumeration: + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_enumerate_oci_registry(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["app/frontend", "app/backend"] + adapter.list_tags.side_effect = [["latest", "v1.0"], ["latest"]] + mock_factory.return_value = adapter + + provider = _build_provider() + assert "myregistry.io/app/frontend:latest" in provider.images + assert "myregistry.io/app/frontend:v1.0" in provider.images + assert "myregistry.io/app/backend:latest" in provider.images + assert len(provider.images) == 3 + + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_image_filter(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["prod/app", "dev/app", "staging/app"] + adapter.list_tags.return_value = ["latest"] + mock_factory.return_value = adapter + + provider = _build_provider(image_filter="^prod/") + assert len(provider.images) == 1 + assert "myregistry.io/prod/app:latest" in provider.images + + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_tag_filter(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["myapp"] + adapter.list_tags.return_value = ["latest", "v1.0", "v2.0", "dev-abc123"] + mock_factory.return_value = adapter + + provider = _build_provider(tag_filter=r"^v\d+\.\d+$") + assert len(provider.images) == 2 + assert "myregistry.io/myapp:v1.0" in provider.images + assert "myregistry.io/myapp:v2.0" in provider.images + + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_combined_filters(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["prod/app", "dev/app"] + adapter.list_tags.return_value = ["latest", "v1.0"] + mock_factory.return_value = adapter + + provider = _build_provider(image_filter="^prod/", tag_filter="^v") + assert len(provider.images) == 1 + assert "myregistry.io/prod/app:v1.0" in provider.images + + +class TestMaxImages: + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_max_images_exceeded(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["app1", "app2", "app3"] + adapter.list_tags.return_value = ["latest", "v1.0"] + mock_factory.return_value = adapter + + with pytest.raises(ImageMaxImagesExceededError): + _build_provider(max_images=2) + + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_max_images_not_exceeded(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["app1"] + adapter.list_tags.return_value = ["latest"] + mock_factory.return_value = adapter + + provider = _build_provider(max_images=10) + assert len(provider.images) == 1 + + +class TestDeduplication: + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_deduplication_with_explicit_images(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["myapp"] + adapter.list_tags.return_value = ["latest"] + mock_factory.return_value = adapter + + provider = _build_provider(images=["myregistry.io/myapp:latest"]) + assert provider.images.count("myregistry.io/myapp:latest") == 1 + + +class TestInvalidFilters: + def test_invalid_image_filter_regex(self): + with pytest.raises(ImageInvalidFilterError): + _build_provider(image_filter="[invalid") + + def test_invalid_tag_filter_regex(self): + with pytest.raises(ImageInvalidFilterError): + _build_provider(tag_filter="(unclosed") + + +class TestRegistryInsecure: + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_insecure_passes_verify_false(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["app"] + adapter.list_tags.return_value = ["latest"] + mock_factory.return_value = adapter + + _build_provider(registry_insecure=True) + mock_factory.assert_called_once() + call_kwargs = mock_factory.call_args[1] + assert call_kwargs["verify_ssl"] is False + + +class TestEmptyRegistry: + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_empty_catalog_with_explicit_images(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = [] + mock_factory.return_value = adapter + + provider = _build_provider(images=["nginx:latest"]) + assert provider.images == ["nginx:latest"] + + +class TestRegistryList: + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_registry_list_prints_and_returns(self, mock_factory, capsys): + adapter = MagicMock() + adapter.list_repositories.return_value = ["app/frontend", "app/backend"] + adapter.list_tags.side_effect = [["latest", "v1.0"], ["latest"]] + mock_factory.return_value = adapter + + provider = _build_provider(registry_list_images=True) + + assert provider._listing_only is True + captured = capsys.readouterr() + assert "app/frontend" in captured.out + assert "app/backend" in captured.out + assert "latest" in captured.out + assert "v1.0" in captured.out + assert "2 repositories" in captured.out + assert "3 images" in captured.out + + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_registry_list_respects_image_filter(self, mock_factory, capsys): + adapter = MagicMock() + adapter.list_repositories.return_value = ["prod/app", "dev/app"] + adapter.list_tags.return_value = ["latest"] + mock_factory.return_value = adapter + + provider = _build_provider(registry_list_images=True, image_filter="^prod/") + + assert provider._listing_only is True + captured = capsys.readouterr() + assert "prod/app" in captured.out + assert "dev/app" not in captured.out + assert "1 repository" in captured.out + + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_registry_list_respects_tag_filter(self, mock_factory, capsys): + adapter = MagicMock() + adapter.list_repositories.return_value = ["myapp"] + adapter.list_tags.return_value = ["latest", "v1.0", "dev-abc"] + mock_factory.return_value = adapter + + provider = _build_provider(registry_list_images=True, tag_filter=r"^v\d+\.\d+$") + + assert provider._listing_only is True + captured = capsys.readouterr() + assert "v1.0" in captured.out + assert "dev-abc" not in captured.out + assert "1 image)" in captured.out + + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_registry_list_skips_max_images(self, mock_factory, capsys): + adapter = MagicMock() + adapter.list_repositories.return_value = ["app1", "app2", "app3"] + adapter.list_tags.return_value = ["latest", "v1.0"] + mock_factory.return_value = adapter + + # max_images=1 would normally raise, but --registry-list skips it + provider = _build_provider(registry_list_images=True, max_images=1) + + assert provider._listing_only is True + captured = capsys.readouterr() + assert "6 images" in captured.out + + +class TestDockerHubEnumeration: + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_dockerhub_images_use_repo_tag_format(self, mock_factory): + """Docker Hub images should use repo:tag format without host prefix.""" + adapter = MagicMock(spec=DockerHubAdapter) + adapter.list_repositories.return_value = ["myorg/app1", "myorg/app2"] + adapter.list_tags.side_effect = [["latest", "v1.0"], ["latest"]] + mock_factory.return_value = adapter + + provider = _build_provider(registry="docker.io/myorg") + # Docker Hub images should NOT have host prefix + assert "myorg/app1:latest" in provider.images + assert "myorg/app1:v1.0" in provider.images + assert "myorg/app2:latest" in provider.images + # Ensure no host prefix was added + for img in provider.images: + assert not img.startswith("docker.io/"), f"Unexpected host prefix in {img}" + assert len(provider.images) == 3