diff --git a/api/CHANGELOG.md b/api/CHANGELOG.md index fee223b251..4c7d1e684e 100644 --- a/api/CHANGELOG.md +++ b/api/CHANGELOG.md @@ -22,6 +22,11 @@ All notable changes to the **Prowler API** are documented in this file. - Attack Paths: Mark attack Paths scan as failed when Celery task fails outside job error handling [(#10065)](https://github.com/prowler-cloud/prowler/pull/10065) - Attack Paths: Remove legacy per-scan `graph_database` and `is_graph_database_deleted` fields from AttackPathsScan model [(#10077)](https://github.com/prowler-cloud/prowler/pull/10077) +### 🐞 Fixed + +- Skip compliance and attack surface tasks for Image and IaC providers +- Normalize Image provider UID by stripping protocol prefix on save + ### 🔐 Security - Bump `Pillow` to 12.1.1 (CVE-2021-25289) [(#10027)](https://github.com/prowler-cloud/prowler/pull/10027) diff --git a/api/src/backend/api/models.py b/api/src/backend/api/models.py index 2efc53bb6a..e794ede1dc 100644 --- a/api/src/backend/api/models.py +++ b/api/src/backend/api/models.py @@ -423,7 +423,7 @@ class Provider(RowLevelSecurityProtectedModel): @staticmethod def validate_image_uid(value): - pattern = r"^(https?://)?[a-zA-Z0-9]([a-zA-Z0-9._-]*[a-zA-Z0-9])?(:\d{1,5})?(/[a-zA-Z0-9._-]+)*/?$" + pattern = r"^[a-zA-Z0-9]([a-zA-Z0-9._-]*[a-zA-Z0-9])?(:\d{1,5})?(/[a-zA-Z0-9._-]+)*/?$" if not re.match(pattern, value): raise ModelValidationError( detail="Image provider ID must be a valid registry URL " @@ -464,6 +464,8 @@ class Provider(RowLevelSecurityProtectedModel): def clean(self): super().clean() + if self.provider == self.ProviderChoices.IMAGE.value and self.uid: + self.uid = re.sub(r"^https?://", "", self.uid) getattr(self, f"validate_{self.provider}_uid")(self.uid) def save(self, *args, **kwargs): diff --git a/api/src/backend/tasks/tasks.py b/api/src/backend/tasks/tasks.py index cbe44ab304..b84b9b7247 100644 --- a/api/src/backend/tasks/tasks.py +++ b/api/src/backend/tasks/tasks.py @@ -10,9 +10,9 @@ from config.django.base import DJANGO_FINDINGS_BATCH_SIZE, DJANGO_TMP_OUTPUT_DIR from django_celery_beat.models import PeriodicTask from tasks.jobs.attack_paths import ( attack_paths_scan, - db_utils as attack_paths_db_utils, can_provider_run_attack_paths_scan, ) +from tasks.jobs.attack_paths import db_utils as attack_paths_db_utils from tasks.jobs.backfill import ( backfill_compliance_summaries, backfill_daily_severity_summaries, @@ -134,13 +134,41 @@ def _perform_scan_complete_tasks(tenant_id: str, scan_id: str, provider_id: str) scan_id (str): The ID of the scan that was performed. provider_id (str): The primary key of the Provider instance that was scanned. """ - chain( - create_compliance_requirements_task.si(tenant_id=tenant_id, scan_id=scan_id), - update_provider_compliance_scores_task.si(tenant_id=tenant_id, scan_id=scan_id), - ).apply_async() - aggregate_attack_surface_task.apply_async( - kwargs={"tenant_id": tenant_id, "scan_id": scan_id} + with rls_transaction(tenant_id): + provider_type = Provider.objects.get(id=provider_id).provider + + has_compliance = provider_type not in ( + Provider.ProviderChoices.IAC.value, + Provider.ProviderChoices.IMAGE.value, ) + + if has_compliance: + chain( + create_compliance_requirements_task.si( + tenant_id=tenant_id, scan_id=scan_id + ), + update_provider_compliance_scores_task.si( + tenant_id=tenant_id, scan_id=scan_id + ), + ).apply_async() + aggregate_attack_surface_task.apply_async( + kwargs={"tenant_id": tenant_id, "scan_id": scan_id} + ) + + final_group_tasks = [ + check_integrations_task.si( + tenant_id=tenant_id, + provider_id=provider_id, + scan_id=scan_id, + ), + ] + if has_compliance: + final_group_tasks.append( + generate_compliance_reports_task.si( + tenant_id=tenant_id, scan_id=scan_id, provider_id=provider_id + ), + ) + chain( perform_scan_summary_task.si(tenant_id=tenant_id, scan_id=scan_id), group( @@ -149,17 +177,7 @@ def _perform_scan_complete_tasks(tenant_id: str, scan_id: str, provider_id: str) scan_id=scan_id, provider_id=provider_id, tenant_id=tenant_id ), ), - group( - # Use optimized task that generates both reports with shared queries - generate_compliance_reports_task.si( - tenant_id=tenant_id, scan_id=scan_id, provider_id=provider_id - ), - check_integrations_task.si( - tenant_id=tenant_id, - provider_id=provider_id, - scan_id=scan_id, - ), - ), + group(*final_group_tasks), ).apply_async() if can_provider_run_attack_paths_scan(tenant_id, provider_id): diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index 2f39207079..23c0609285 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -40,6 +40,10 @@ All notable changes to the **Prowler SDK** are documented in this file. - Update GCP KMS service metadata to new format [(#9647)](https://github.com/prowler-cloud/prowler/pull/9647) - Update GCP Logging service metadata to new format [(#9648)](https://github.com/prowler-cloud/prowler/pull/9648) +### 🐞 Fixed + +- `image` provider: dynamic User-Agent version, correct finding Categories and RelatedUrl metadata, remove dead docker login/pull code, replace SystemExit in registry listing + ### 🔐 Security - Bumped `py-ocsf-models` to 0.8.1 and `cryptography` to 44.0.3 [(#10059)](https://github.com/prowler-cloud/prowler/pull/10059) diff --git a/prowler/providers/image/exceptions/exceptions.py b/prowler/providers/image/exceptions/exceptions.py index 509a3f1c3e..387b443ce3 100644 --- a/prowler/providers/image/exceptions/exceptions.py +++ b/prowler/providers/image/exceptions/exceptions.py @@ -50,14 +50,6 @@ class ImageBaseException(ProwlerException): "message": "Invalid image config scanner type.", "remediation": "Use valid image config scanners: misconfig, secret.", }, - (11011, "ImageDockerLoginError"): { - "message": "Docker login failed for registry authentication.", - "remediation": "Check your registry credentials and ensure the registry is reachable.", - }, - (11012, "ImageDockerNotFoundError"): { - "message": "Docker binary not found.", - "remediation": "Install Docker to enable private registry authentication via docker login.", - }, (11013, "ImageRegistryAuthError"): { "message": "Registry authentication failed.", "remediation": "Check REGISTRY_USERNAME/REGISTRY_PASSWORD or REGISTRY_TOKEN environment variables.", @@ -192,24 +184,6 @@ class ImageInvalidConfigScannerError(ImageBaseException): ) -class ImageDockerLoginError(ImageBaseException): - """Exception raised when docker login fails.""" - - def __init__(self, file=None, original_exception=None, message=None): - super().__init__( - 11011, file=file, original_exception=original_exception, message=message - ) - - -class ImageDockerNotFoundError(ImageBaseException): - """Exception raised when the docker binary is not found.""" - - def __init__(self, file=None, original_exception=None, message=None): - super().__init__( - 11012, file=file, original_exception=original_exception, message=message - ) - - class ImageRegistryAuthError(ImageBaseException): """Exception raised when registry authentication fails.""" diff --git a/prowler/providers/image/image_provider.py b/prowler/providers/image/image_provider.py index 6a5bdc34f3..41d499b122 100644 --- a/prowler/providers/image/image_provider.py +++ b/prowler/providers/image/image_provider.py @@ -96,6 +96,7 @@ class ImageProvider(Provider): 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( @@ -152,6 +153,8 @@ class ImageProvider(Provider): # 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) @@ -337,22 +340,25 @@ class ImageProvider(Provider): CheckReportImage: The processed check report """ try: - # Determine finding ID based on type + # 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 = "" @@ -381,7 +387,7 @@ class ImageProvider(Provider): "Risk": finding.get( "Description", "Vulnerability detected in container image" ), - "RelatedUrl": finding.get("PrimaryURL", ""), + "RelatedUrl": "", "Remediation": { "Code": { "NativeIaC": "", @@ -394,7 +400,7 @@ class ImageProvider(Provider): "Url": finding.get("PrimaryURL", ""), }, }, - "Categories": [], + "Categories": finding_categories, "DependsOn": [], "RelatedTo": [], "Notes": "", @@ -770,10 +776,11 @@ class ImageProvider(Provider): image_ref = f"{registry_host}/{repo}:{tag}" discovered_images.append(image_ref) - # Registry list mode: print listing and exit + # Registry list mode: print listing and return early if self.registry_list_images: self._print_registry_listing(repos_tags, len(discovered_images)) - raise SystemExit(0) + self._listing_only = True + return # Check max-images limit if self.max_images and len(discovered_images) > self.max_images: diff --git a/prowler/providers/image/lib/registry/base.py b/prowler/providers/image/lib/registry/base.py index 5582f217bb..133ef9d4e0 100644 --- a/prowler/providers/image/lib/registry/base.py +++ b/prowler/providers/image/lib/registry/base.py @@ -8,12 +8,13 @@ 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 = "Prowler/1.0 (registry-adapter)" +_USER_AGENT = f"Prowler/{prowler_version} (registry-adapter)" class RegistryAdapter(ABC): diff --git a/tests/lib/scan/scan_test.py b/tests/lib/scan/scan_test.py index e6ff948ed7..93cae2f55f 100644 --- a/tests/lib/scan/scan_test.py +++ b/tests/lib/scan/scan_test.py @@ -416,3 +416,97 @@ class TestScan: results = list(scan.scan(custom_checks_metadata)) assert results[0] == (100.0, []) + + +class TestImageScanPath: + """Tests for the IMAGE provider scan path in Scan.scan().""" + + @staticmethod + def _make_check_metadata(check_id): + from prowler.lib.check.models import CheckMetadata + + return CheckMetadata( + Provider="image", + CheckID=check_id, + CheckTitle=f"Test {check_id}", + CheckType=["Container Image Security"], + ServiceName="vuln", + SubServiceName="", + ResourceIdTemplate="", + Severity="high", + ResourceType="container-image", + ResourceGroup="container", + Description="Test finding", + Risk="Test risk", + RelatedUrl="", + Remediation={ + "Code": {"NativeIaC": "", "Terraform": "", "CLI": "", "Other": ""}, + "Recommendation": {"Text": "", "Url": ""}, + }, + Categories=["vulnerability"], + DependsOn=[], + RelatedTo=[], + Notes="", + ) + + @staticmethod + def _make_report(check_id, image_name, status="FAIL"): + report = MagicMock() + report.check_metadata = TestImageScanPath._make_check_metadata(check_id) + report.resource_name = image_name + report.resource_id = f"{image_name}-resource" + report.status = status + report.status_extended = f"{check_id} found" + report.muted = False + report.resource = {} + report.resource_details = "" + report.region = "container" + return report + + @pytest.fixture + def mock_image_provider(self): + """Create a mock ImageProvider that passes isinstance checks.""" + from prowler.providers.image.image_provider import ImageProvider + + provider = MagicMock(spec=ImageProvider) + provider.type = "image" + provider.images = ["img1:latest", "img2:latest"] + provider.registry = None + + r1 = self._make_report("CVE-2024-0001", "img1:latest") + r2 = self._make_report("CVE-2024-0002", "img2:latest") + provider.scan_per_image.return_value = iter( + [("img1:latest", [r1]), ("img2:latest", [r2])] + ) + return provider + + def test_image_scan_yields_findings(self, mock_image_provider): + """Verify Finding objects are created with correct fields for each image.""" + scan = Scan(mock_image_provider) + results = list(scan.scan({})) + + assert len(results) == 2 + for _, findings in results: + assert len(findings) == 1 + f = findings[0] + assert f.status.name in ("PASS", "FAIL") + assert f.auth_method == "Registry" + assert f.account_name == "Container Registry" + + def test_image_scan_progress_tracking(self, mock_image_provider): + """Verify progress increments per image.""" + scan = Scan(mock_image_provider) + results = list(scan.scan({})) + + progresses = [p for p, _ in results] + assert len(progresses) == 2 + assert progresses[0] < progresses[1] + + def test_image_scan_status_filtering(self, mock_image_provider): + """Verify status filter excludes non-matching findings.""" + scan = Scan(mock_image_provider, status=["PASS"]) + results = list(scan.scan({})) + + # All reports have status="FAIL", so filtering for PASS should yield empty findings + for _, findings in results: + assert len(findings) == 0 diff --git a/tests/providers/image/image_provider_test.py b/tests/providers/image/image_provider_test.py index 5e3f7acbed..59ae20ce34 100644 --- a/tests/providers/image/image_provider_test.py +++ b/tests/providers/image/image_provider_test.py @@ -7,8 +7,6 @@ import pytest from prowler.lib.check.models import CheckReportImage from prowler.providers.image.exceptions.exceptions import ( - ImageDockerLoginError, - ImageDockerNotFoundError, ImageInvalidConfigScannerError, ImageInvalidNameError, ImageInvalidScannerError, @@ -138,6 +136,8 @@ class TestImageProvider: assert report.fixed_version == "1.1.1l-r0" assert report.resource_name == "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).""" @@ -153,6 +153,7 @@ class TestImageProvider: assert report.check_metadata.CheckID == "aws-access-key-id" assert report.check_metadata.Severity == "critical" assert report.check_metadata.ServiceName == "secret" + assert report.check_metadata.Categories == ["secrets"] def test_process_finding_misconfiguration(self): """Test processing a misconfiguration finding (identified by ID).""" @@ -167,6 +168,7 @@ class TestImageProvider: assert report.check_metadata.CheckID == "DS001" assert report.check_metadata.Severity == "medium" assert report.check_metadata.ServiceName == "misconfiguration" + assert report.check_metadata.Categories == [] def test_process_finding_unknown_severity(self): """Test that UNKNOWN severity is mapped to informational.""" @@ -486,16 +488,16 @@ class TestImageProviderRegistryAuth: assert "TRIVY_PASSWORD" not in env assert "TRIVY_REGISTRY_TOKEN" not in env - def test_build_trivy_env_basic_auth_no_env_vars(self): - """Test that _build_trivy_env does NOT inject TRIVY_USERNAME/PASSWORD (docker login handles it).""" + 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 "TRIVY_USERNAME" not in env - assert "TRIVY_PASSWORD" not in 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.""" @@ -505,8 +507,8 @@ class TestImageProviderRegistryAuth: assert env["TRIVY_REGISTRY_TOKEN"] == "my-token" @patch("subprocess.run") - def test_execute_trivy_no_trivy_env_with_basic_auth(self, mock_subprocess): - """Test that _execute_trivy does NOT set TRIVY_USERNAME/PASSWORD (docker login handles auth).""" + 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", @@ -519,12 +521,12 @@ class TestImageProviderRegistryAuth: call_kwargs = mock_subprocess.call_args env = call_kwargs.kwargs.get("env") or call_kwargs[1].get("env") - assert "TRIVY_USERNAME" not in env - assert "TRIVY_PASSWORD" not in 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 does docker login, pull, trivy, logout and does NOT set TRIVY_* env vars.""" + """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( @@ -534,29 +536,12 @@ class TestImageProviderRegistryAuth: ) assert result.is_connected is True - # Should have 4 subprocess calls: docker login, docker pull, trivy, docker logout - assert mock_subprocess.call_count == 4 - login_call = mock_subprocess.call_args_list[0] - assert login_call.args[0] == [ - "docker", - "login", - "--username", - "myuser", - "--password-stdin", - "private.registry.io", - ] - assert login_call.kwargs["input"] == "mypass" - - pull_call = mock_subprocess.call_args_list[1] - assert pull_call.args[0] == ["docker", "pull", "private.registry.io/myapp:v1"] - - trivy_call = mock_subprocess.call_args_list[2] + 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 "TRIVY_USERNAME" not in env - assert "TRIVY_PASSWORD" not in env - - logout_call = mock_subprocess.call_args_list[3] - assert logout_call.args[0] == ["docker", "logout", "private.registry.io"] + assert env["TRIVY_USERNAME"] == "myuser" + assert env["TRIVY_PASSWORD"] == "mypass" @patch("subprocess.run") def test_test_connection_with_token(self, mock_subprocess): @@ -627,159 +612,7 @@ class TestExtractRegistry: assert ImageProvider._extract_registry("nginx") is None -class TestDockerLogin: - @patch("subprocess.run") - def test_login_success(self, mock_subprocess): - """Test successful docker login.""" - mock_subprocess.return_value = MagicMock(returncode=0, stderr="") - provider = _make_provider( - registry_username="myuser", - registry_password="mypass", - ) - - provider._docker_login("ghcr.io") - - mock_subprocess.assert_called_once() - call_args = mock_subprocess.call_args - assert call_args.args[0] == [ - "docker", - "login", - "--username", - "myuser", - "--password-stdin", - "ghcr.io", - ] - assert call_args.kwargs["input"] == "mypass" - assert "ghcr.io" in provider._logged_in_registries - - @patch("subprocess.run") - def test_login_docker_hub(self, mock_subprocess): - """Test docker login for Docker Hub (registry=None).""" - mock_subprocess.return_value = MagicMock(returncode=0, stderr="") - provider = _make_provider( - registry_username="myuser", - registry_password="mypass", - ) - - provider._docker_login(None) - - call_args = mock_subprocess.call_args - assert call_args.args[0] == [ - "docker", - "login", - "--username", - "myuser", - "--password-stdin", - ] - assert None in provider._logged_in_registries - - @patch("subprocess.run") - def test_login_failure_raises(self, mock_subprocess): - """Test docker login failure raises ImageDockerLoginError.""" - mock_subprocess.return_value = MagicMock( - returncode=1, stderr="unauthorized: incorrect username or password" - ) - provider = _make_provider( - registry_username="myuser", - registry_password="badpass", - ) - - with pytest.raises(ImageDockerLoginError): - provider._docker_login("ghcr.io") - - assert "ghcr.io" not in provider._logged_in_registries - - @patch("subprocess.run") - def test_login_docker_not_found(self, mock_subprocess): - """Test docker binary not found raises ImageDockerNotFoundError.""" - mock_subprocess.side_effect = FileNotFoundError("docker not found") - provider = _make_provider( - registry_username="myuser", - registry_password="mypass", - ) - - with pytest.raises(ImageDockerNotFoundError): - provider._docker_login("ghcr.io") - - @patch("subprocess.run") - def test_login_password_via_stdin_not_args(self, mock_subprocess): - """Test that password is passed via stdin, never in command args.""" - mock_subprocess.return_value = MagicMock(returncode=0, stderr="") - provider = _make_provider( - registry_username="myuser", - registry_password="s3cret!", - ) - - provider._docker_login("ghcr.io") - - call_args = mock_subprocess.call_args - cmd = call_args.args[0] - assert "s3cret!" not in cmd - assert call_args.kwargs["input"] == "s3cret!" - - -class TestDockerLogout: - @patch("subprocess.run") - def test_logout_success(self, mock_subprocess): - """Test successful docker logout.""" - mock_subprocess.return_value = MagicMock(returncode=0, stderr="") - provider = _make_provider() - - provider._docker_logout("ghcr.io") - - mock_subprocess.assert_called_once_with( - ["docker", "logout", "ghcr.io"], - capture_output=True, - text=True, - ) - - @patch("subprocess.run") - def test_logout_docker_hub(self, mock_subprocess): - """Test docker logout for Docker Hub (registry=None).""" - mock_subprocess.return_value = MagicMock(returncode=0, stderr="") - provider = _make_provider() - - provider._docker_logout(None) - - mock_subprocess.assert_called_once_with( - ["docker", "logout"], - capture_output=True, - text=True, - ) - - @patch("subprocess.run") - def test_logout_failure_warns(self, mock_subprocess): - """Test docker logout failure only warns, never raises.""" - mock_subprocess.return_value = MagicMock(returncode=1, stderr="some error") - provider = _make_provider() - - provider._docker_logout("ghcr.io") - - @patch("subprocess.run") - def test_logout_exception_warns(self, mock_subprocess): - """Test docker logout handles exceptions gracefully.""" - mock_subprocess.side_effect = OSError("docker crashed") - provider = _make_provider() - - provider._docker_logout("ghcr.io") - - class TestCleanup: - @patch("subprocess.run") - def test_cleanup_calls_logout_for_each_registry(self, mock_subprocess): - """Test cleanup calls docker logout for each logged-in registry.""" - mock_subprocess.return_value = MagicMock(returncode=0, stderr="") - provider = _make_provider( - registry_username="myuser", - registry_password="mypass", - ) - provider._logged_in_registries = {"ghcr.io", None} - - provider.cleanup() - - assert mock_subprocess.call_count == 2 - assert len(provider._logged_in_registries) == 0 - def test_cleanup_idempotent(self): """Test cleanup is safe to call multiple times.""" provider = _make_provider() @@ -787,217 +620,6 @@ class TestCleanup: provider.cleanup() provider.cleanup() - assert len(provider._logged_in_registries) == 0 - - @patch("subprocess.run") - def test_cleanup_clears_set(self, mock_subprocess): - """Test cleanup clears the _logged_in_registries set.""" - mock_subprocess.return_value = MagicMock(returncode=0, stderr="") - provider = _make_provider( - registry_username="myuser", - registry_password="mypass", - ) - provider._logged_in_registries = {"ghcr.io"} - - provider.cleanup() - - assert provider._logged_in_registries == set() - - -class TestDockerPull: - @patch("subprocess.run") - def test_pull_success(self, mock_subprocess): - """Test successful docker pull.""" - mock_subprocess.return_value = MagicMock(returncode=0, stderr="") - provider = _make_provider( - registry_username="myuser", - registry_password="mypass", - ) - - provider._docker_pull("ghcr.io/user/image:tag") - - mock_subprocess.assert_called_once_with( - ["docker", "pull", "ghcr.io/user/image:tag"], - capture_output=True, - text=True, - ) - - @patch("subprocess.run") - def test_pull_failure_raises_scan_error(self, mock_subprocess): - """Test docker pull failure raises ImageScanError.""" - mock_subprocess.return_value = MagicMock( - returncode=1, stderr="Error: pull access denied" - ) - provider = _make_provider( - registry_username="myuser", - registry_password="mypass", - ) - - with pytest.raises(ImageScanError): - provider._docker_pull("private/image:tag") - - @patch("subprocess.run") - def test_pull_docker_not_found(self, mock_subprocess): - """Test docker binary not found raises ImageDockerNotFoundError.""" - mock_subprocess.side_effect = FileNotFoundError("docker not found") - provider = _make_provider( - registry_username="myuser", - registry_password="mypass", - ) - - with pytest.raises(ImageDockerNotFoundError): - provider._docker_pull("alpine:3.18") - - -class TestDockerLoginIntegration: - @patch("subprocess.run") - def test_run_scan_calls_docker_login_with_credentials(self, mock_subprocess): - """Test that run_scan() calls docker login, docker pull, then trivy when credentials are set.""" - mock_subprocess.return_value = MagicMock( - returncode=0, stdout=get_sample_trivy_json_output(), stderr="" - ) - provider = _make_provider( - images=["ghcr.io/user/image:tag"], - registry_username="myuser", - registry_password="mypass", - ) - - reports = [] - for batch in provider.run_scan(): - reports.extend(batch) - - calls = mock_subprocess.call_args_list - assert calls[0].args[0][:2] == ["docker", "login"] - assert calls[1].args[0][:2] == ["docker", "pull"] - assert calls[2].args[0][:2] == ["trivy", "image"] - - @patch("subprocess.run") - def test_run_scan_no_docker_login_without_credentials(self, mock_subprocess): - """Test that run_scan() does NOT call docker login when no credentials.""" - mock_subprocess.return_value = MagicMock( - returncode=0, stdout=get_sample_trivy_json_output(), stderr="" - ) - provider = _make_provider() - - for batch in provider.run_scan(): - pass - - calls = mock_subprocess.call_args_list - assert all(call.args[0][0] == "trivy" for call in calls) - - @patch("subprocess.run") - def test_run_scan_no_docker_login_with_token_only(self, mock_subprocess): - """Test that run_scan() does NOT call docker login when only token is provided.""" - mock_subprocess.return_value = MagicMock( - returncode=0, stdout=get_sample_trivy_json_output(), stderr="" - ) - provider = _make_provider(registry_token="my-token") - - for batch in provider.run_scan(): - pass - - calls = mock_subprocess.call_args_list - assert all(call.args[0][0] == "trivy" for call in calls) - - @patch("subprocess.run") - def test_run_calls_cleanup_on_success(self, mock_subprocess): - """Test that run() calls cleanup after successful scan.""" - mock_subprocess.return_value = MagicMock( - returncode=0, stdout=get_sample_trivy_json_output(), stderr="" - ) - provider = _make_provider( - images=["ghcr.io/user/image:tag"], - registry_username="myuser", - registry_password="mypass", - ) - - provider.run() - - calls = mock_subprocess.call_args_list - # login, pull, trivy, logout - assert calls[0].args[0][:2] == ["docker", "login"] - assert calls[1].args[0][:2] == ["docker", "pull"] - assert calls[2].args[0][:2] == ["trivy", "image"] - assert calls[-1].args[0][:2] == ["docker", "logout"] - - @patch("subprocess.run") - def test_run_calls_cleanup_on_error(self, mock_subprocess): - """Test that run() calls cleanup even when scan errors.""" - - def side_effect(*args, **kwargs): - cmd = args[0] - if cmd[0] == "docker" and cmd[1] == "login": - return MagicMock(returncode=0, stderr="") - if cmd[0] == "docker" and cmd[1] == "pull": - return MagicMock(returncode=0, stderr="") - if cmd[0] == "trivy": - return MagicMock(returncode=1, stdout="", stderr="scan failed") - return MagicMock(returncode=0, stderr="") - - mock_subprocess.side_effect = side_effect - provider = _make_provider( - images=["ghcr.io/user/image:tag"], - registry_username="myuser", - registry_password="mypass", - ) - - with pytest.raises(ImageScanError): - provider.run() - - last_call = mock_subprocess.call_args_list[-1] - assert last_call.args[0][:2] == ["docker", "logout"] - - @patch("subprocess.run") - def test_run_scan_deduplicates_registries(self, mock_subprocess): - """Test that run_scan() deduplicates registries for docker login but pulls each image.""" - mock_subprocess.return_value = MagicMock( - returncode=0, stdout=get_sample_trivy_json_output(), stderr="" - ) - provider = _make_provider( - images=["ghcr.io/user/image1:tag", "ghcr.io/user/image2:tag"], - registry_username="myuser", - registry_password="mypass", - ) - - for batch in provider.run_scan(): - pass - - calls = mock_subprocess.call_args_list - docker_login_calls = [c for c in calls if c.args[0][:2] == ["docker", "login"]] - docker_pull_calls = [c for c in calls if c.args[0][:2] == ["docker", "pull"]] - trivy_calls = [c for c in calls if c.args[0][:2] == ["trivy", "image"]] - assert len(docker_login_calls) == 1 - assert len(docker_pull_calls) == 2 - assert len(trivy_calls) == 2 - - @patch("subprocess.run") - def test_test_connection_docker_login_docker_hub(self, mock_subprocess): - """Test test_connection does docker login and pull for Docker Hub images.""" - mock_subprocess.return_value = MagicMock(returncode=0, stderr="") - - result = ImageProvider.test_connection( - image="andoniaf/test-private:tag", - registry_username="myuser", - registry_password="mypass", - ) - - assert result.is_connected is True - assert mock_subprocess.call_count == 4 - login_call = mock_subprocess.call_args_list[0] - assert login_call.args[0] == [ - "docker", - "login", - "--username", - "myuser", - "--password-stdin", - ] - pull_call = mock_subprocess.call_args_list[1] - assert pull_call.args[0] == [ - "docker", - "pull", - "andoniaf/test-private:tag", - ] - class TestImageProviderInputValidation: def test_invalid_timeout_format_raises_error(self): @@ -1188,3 +810,67 @@ class TestImageProviderNameValidation: 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/registry/test_dockerhub_adapter.py b/tests/providers/image/lib/registry/test_dockerhub_adapter.py index 5334ea083e..930e872bba 100644 --- a/tests/providers/image/lib/registry/test_dockerhub_adapter.py +++ b/tests/providers/image/lib/registry/test_dockerhub_adapter.py @@ -197,7 +197,12 @@ class TestDockerHubRetry: adapter = DockerHubAdapter("docker.io/myorg") adapter._request_with_retry("GET", "https://hub.docker.com") _, kwargs = mock_request.call_args - assert kwargs["headers"]["User-Agent"] == "Prowler/1.0 (registry-adapter)" + 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") diff --git a/tests/providers/image/lib/registry/test_provider_registry.py b/tests/providers/image/lib/registry/test_provider_registry.py index 86667f0412..97901ea0c4 100644 --- a/tests/providers/image/lib/registry/test_provider_registry.py +++ b/tests/providers/image/lib/registry/test_provider_registry.py @@ -152,16 +152,15 @@ class TestEmptyRegistry: class TestRegistryList: @patch("prowler.providers.image.image_provider.create_registry_adapter") - def test_registry_list_prints_and_exits(self, mock_factory, capsys): + 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 - with pytest.raises(SystemExit) as exc_info: - _build_provider(registry_list_images=True) + provider = _build_provider(registry_list_images=True) - assert exc_info.value.code == 0 + assert provider._listing_only is True captured = capsys.readouterr() assert "app/frontend" in captured.out assert "app/backend" in captured.out @@ -177,10 +176,9 @@ class TestRegistryList: adapter.list_tags.return_value = ["latest"] mock_factory.return_value = adapter - with pytest.raises(SystemExit) as exc_info: - _build_provider(registry_list_images=True, image_filter="^prod/") + provider = _build_provider(registry_list_images=True, image_filter="^prod/") - assert exc_info.value.code == 0 + assert provider._listing_only is True captured = capsys.readouterr() assert "prod/app" in captured.out assert "dev/app" not in captured.out @@ -193,10 +191,9 @@ class TestRegistryList: adapter.list_tags.return_value = ["latest", "v1.0", "dev-abc"] mock_factory.return_value = adapter - with pytest.raises(SystemExit) as exc_info: - _build_provider(registry_list_images=True, tag_filter=r"^v\d+\.\d+$") + provider = _build_provider(registry_list_images=True, tag_filter=r"^v\d+\.\d+$") - assert exc_info.value.code == 0 + assert provider._listing_only is True captured = capsys.readouterr() assert "v1.0" in captured.out assert "dev-abc" not in captured.out @@ -210,10 +207,9 @@ class TestRegistryList: mock_factory.return_value = adapter # max_images=1 would normally raise, but --registry-list skips it - with pytest.raises(SystemExit) as exc_info: - _build_provider(registry_list_images=True, max_images=1) + provider = _build_provider(registry_list_images=True, max_images=1) - assert exc_info.value.code == 0 + assert provider._listing_only is True captured = capsys.readouterr() assert "6 images" in captured.out