mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-26 13:24:20 +00:00
fix(image): add input validation and file size limits
- Add IMAGE to Provider enum for runtime routing - Validate image names against shell metacharacters and OCI format - Enforce max line count and name length on image list files - Add tests for name validation and file size limits
This commit is contained in:
4 files changed
+124
-2
No files matched your search
@@ -63,6 +63,7 @@ class Provider(str, Enum):
|
||||
ORACLECLOUD = "oraclecloud"
|
||||
ALIBABACLOUD = "alibabacloud"
|
||||
OPENSTACK = "openstack"
|
||||
IMAGE = "image"
|
||||
|
||||
|
||||
# Compliance
|
||||
|
||||
@@ -42,6 +42,10 @@ class ImageBaseException(ProwlerException):
|
||||
"message": "Invalid severity level.",
|
||||
"remediation": "Use valid severities: CRITICAL, HIGH, MEDIUM, LOW, UNKNOWN.",
|
||||
},
|
||||
(9009, "ImageInvalidNameError"): {
|
||||
"message": "Invalid container image name.",
|
||||
"remediation": "Use a valid image reference (e.g., 'alpine:3.18', 'registry.example.com/repo/image:tag').",
|
||||
},
|
||||
}
|
||||
|
||||
def __init__(self, code, file=None, original_exception=None, message=None):
|
||||
@@ -136,3 +140,12 @@ class ImageInvalidSeverityError(ImageBaseException):
|
||||
super().__init__(
|
||||
9008, 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__(
|
||||
9009, file=file, original_exception=original_exception, message=message
|
||||
)
|
||||
@@ -20,6 +20,7 @@ from prowler.providers.common.models import Audit_Metadata, Connection
|
||||
from prowler.providers.common.provider import Provider
|
||||
from prowler.providers.image.exceptions.exceptions import (
|
||||
ImageFindingProcessingError,
|
||||
ImageInvalidNameError,
|
||||
ImageInvalidScannerError,
|
||||
ImageInvalidSeverityError,
|
||||
ImageInvalidTimeoutError,
|
||||
@@ -45,6 +46,10 @@ class ImageProvider(Provider):
|
||||
|
||||
_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__(
|
||||
@@ -79,6 +84,9 @@ class ImageProvider(Provider):
|
||||
if image_list_file:
|
||||
self._load_images_from_file(image_list_file)
|
||||
|
||||
for image in self.images:
|
||||
self._validate_image_name(image)
|
||||
|
||||
if not self.images:
|
||||
raise ImageNoImagesProvidedError(
|
||||
file=__file__,
|
||||
@@ -115,17 +123,32 @@ class ImageProvider(Provider):
|
||||
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 line and not line.startswith("#"):
|
||||
self.images.append(line)
|
||||
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,
|
||||
@@ -155,6 +178,32 @@ class ImageProvider(Provider):
|
||||
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
|
||||
|
||||
@@ -6,10 +6,12 @@ import pytest
|
||||
|
||||
from prowler.lib.check.models import CheckReportImage
|
||||
from prowler.providers.image.exceptions.exceptions import (
|
||||
ImageInvalidNameError,
|
||||
ImageInvalidScannerError,
|
||||
ImageInvalidSeverityError,
|
||||
ImageInvalidTimeoutError,
|
||||
ImageListFileNotFoundError,
|
||||
ImageListFileReadError,
|
||||
ImageNoImagesProvidedError,
|
||||
ImageScanError,
|
||||
ImageTrivyBinaryNotFoundError,
|
||||
@@ -485,3 +487,60 @@ class TestImageProviderErrorCategorization:
|
||||
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)
|
||||
Reference in new issue
Block a user