fix(image): address PR review items for image provider

- Remove dead docker login/pull/logout code and exceptions
- Fix finding metadata: empty RelatedUrl, Categories per finding type
- Replace SystemExit(0) with _listing_only flag in registry listing
- Use dynamic prowler_version in User-Agent header
- Normalize image UID by stripping protocol prefix on save
- Skip compliance and attack surface tasks for Image and IaC providers
- Add tests for scan_per_image and IMAGE scan path

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)

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):

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):

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)

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."""
 

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:

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):

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

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()

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")

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