feat(image): add image provider API support

- Add database migration, model choice, and UID validation for image provider
- Add ImageProviderSecret serializer with registry auth and filter fields
- Add provider kwargs builder and connection test via registry adapter
- Skip compliance and attack surface tasks for image provider scans
- Add scan_per_image generator, SHA extraction, and listing_only mode to SDK
- Update scan class to handle image provider like IaC (Trivy-based)
- Bump Trivy from 0.66.0 to 0.69.1

api/src/backend/api/migrations/0081_image_provider.py

@@ -0,0 +1,40 @@
+# Generated by Django migration for Image provider support
+
+from django.db import migrations
+
+import api.db_utils
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0080_backfill_attack_paths_graph_data_ready"),
+    ]
+
+    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"),
+                    ("openstack", "OpenStack"),
+                    ("image", "Image"),
+                ],
+                default="aws",
+            ),
+        ),
+        migrations.RunSQL(
+            "ALTER TYPE provider ADD VALUE IF NOT EXISTS 'image';",
+            reverse_sql=migrations.RunSQL.noop,
+        ),
+    ]

api/src/backend/api/models.py

@@ -289,6 +289,7 @@ class Provider(RowLevelSecurityProtectedModel):
         ALIBABACLOUD = "alibabacloud", _("Alibaba Cloud")
         CLOUDFLARE = "cloudflare", _("Cloudflare")
         OPENSTACK = "openstack", _("OpenStack")
+        IMAGE = "image", _("Image")
 
     @staticmethod
     def validate_aws_uid(value):
@@ -423,6 +424,27 @@ class Provider(RowLevelSecurityProtectedModel):
                 pointer="/data/attributes/uid",
             )
 
+    @staticmethod
+    def validate_image_uid(value):
+        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 (e.g., ghcr.io, docker.io, "
+                "123456789012.dkr.ecr.us-east-1.amazonaws.com).",
+                code="image-uid",
+                pointer="/data/attributes/uid",
+            )
+        # Validate port range if present
+        port_match = re.search(r":(\d{1,5})(?=/|$)", value)
+        if port_match:
+            port = int(port_match.group(1))
+            if port < 1 or port > 65535:
+                raise ModelValidationError(
+                    detail="Image provider registry port must be between 1 and 65535.",
+                    code="image-uid",
+                    pointer="/data/attributes/uid",
+                )
+
     id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
     inserted_at = models.DateTimeField(auto_now_add=True, editable=False)
     updated_at = models.DateTimeField(auto_now=True, editable=False)
@@ -446,6 +468,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/api/tests/test_utils.py

@@ -24,6 +24,7 @@ from prowler.providers.cloudflare.cloudflare_provider import CloudflareProvider
 from prowler.providers.gcp.gcp_provider import GcpProvider
 from prowler.providers.github.github_provider import GithubProvider
 from prowler.providers.iac.iac_provider import IacProvider
+from prowler.providers.image.image_provider import ImageProvider
 from prowler.providers.kubernetes.kubernetes_provider import KubernetesProvider
 from prowler.providers.m365.m365_provider import M365Provider
 from prowler.providers.mongodbatlas.mongodbatlas_provider import MongodbatlasProvider
@@ -122,6 +123,7 @@ class TestReturnProwlerProvider:
             (Provider.ProviderChoices.ALIBABACLOUD.value, AlibabacloudProvider),
             (Provider.ProviderChoices.CLOUDFLARE.value, CloudflareProvider),
             (Provider.ProviderChoices.OPENSTACK.value, OpenstackProvider),
+            (Provider.ProviderChoices.IMAGE.value, ImageProvider),
         ],
     )
     def test_return_prowler_provider(self, provider_type, expected_provider):
@@ -188,6 +190,54 @@ class TestProwlerProviderConnectionTest:
         assert isinstance(connection.error, Provider.secret.RelatedObjectDoesNotExist)
         assert str(connection.error) == "Provider has no secret."
 
+    @patch(
+        "prowler.providers.image.lib.registry.factory.create_registry_adapter",
+    )
+    @patch("api.utils.return_prowler_provider")
+    def test_prowler_provider_connection_test_image_success(
+        self, mock_return_prowler_provider, mock_create_adapter
+    ):
+        provider = MagicMock()
+        provider.provider = Provider.ProviderChoices.IMAGE.value
+        provider.uid = "ghcr.io"
+        provider.secret.secret = {"registry_token": "tok"}
+
+        mock_adapter = MagicMock()
+        mock_create_adapter.return_value = mock_adapter
+
+        connection = prowler_provider_connection_test(provider)
+
+        assert connection.is_connected is True
+        assert connection.error is None
+        mock_create_adapter.assert_called_once_with(
+            registry_url="ghcr.io",
+            username=None,
+            password=None,
+            token="tok",
+        )
+        mock_adapter.list_repositories.assert_called_once()
+
+    @patch(
+        "prowler.providers.image.lib.registry.factory.create_registry_adapter",
+    )
+    @patch("api.utils.return_prowler_provider")
+    def test_prowler_provider_connection_test_image_failure(
+        self, mock_return_prowler_provider, mock_create_adapter
+    ):
+        provider = MagicMock()
+        provider.provider = Provider.ProviderChoices.IMAGE.value
+        provider.uid = "ghcr.io"
+        provider.secret.secret = {"registry_token": "bad-token"}
+
+        mock_adapter = MagicMock()
+        mock_adapter.list_repositories.side_effect = Exception("401 Unauthorized")
+        mock_create_adapter.return_value = mock_adapter
+
+        connection = prowler_provider_connection_test(provider)
+
+        assert connection.is_connected is False
+        assert connection.error == "401 Unauthorized"
+
 
 class TestGetProwlerProviderKwargs:
     @pytest.mark.parametrize(
@@ -336,6 +386,103 @@ class TestGetProwlerProviderKwargs:
         }
         assert result == expected_result
 
+    def test_get_prowler_provider_kwargs_image_provider(self):
+        """Test that Image provider gets correct kwargs with registry URL and auth."""
+        provider_uid = "ghcr.io"
+        secret_dict = {
+            "registry_username": "user",
+            "registry_password": "pass",
+        }
+        secret_mock = MagicMock()
+        secret_mock.secret = secret_dict
+
+        provider = MagicMock()
+        provider.provider = Provider.ProviderChoices.IMAGE.value
+        provider.secret = secret_mock
+        provider.uid = provider_uid
+
+        result = get_prowler_provider_kwargs(provider)
+
+        expected_result = {
+            "registry": provider_uid,
+            "registry_username": "user",
+            "registry_password": "pass",
+        }
+        assert result == expected_result
+
+    def test_get_prowler_provider_kwargs_image_provider_with_filters(self):
+        """Test that Image provider includes scan filters."""
+        provider_uid = "docker.io"
+        secret_dict = {
+            "registry_token": "ghp_abc123",
+            "image_filter": "my-app.*",
+            "tag_filter": "v[0-9]+",
+            "max_images": 50,
+        }
+        secret_mock = MagicMock()
+        secret_mock.secret = secret_dict
+
+        provider = MagicMock()
+        provider.provider = Provider.ProviderChoices.IMAGE.value
+        provider.secret = secret_mock
+        provider.uid = provider_uid
+
+        result = get_prowler_provider_kwargs(provider)
+
+        expected_result = {
+            "registry": provider_uid,
+            "registry_token": "ghp_abc123",
+            "image_filter": "my-app.*",
+            "tag_filter": "v[0-9]+",
+            "max_images": 50,
+        }
+        assert result == expected_result
+
+    def test_get_prowler_provider_kwargs_image_provider_no_auth(self):
+        """Test that Image provider works with empty secret for public registries."""
+        provider_uid = "docker.io"
+        secret_dict = {}
+        secret_mock = MagicMock()
+        secret_mock.secret = secret_dict
+
+        provider = MagicMock()
+        provider.provider = Provider.ProviderChoices.IMAGE.value
+        provider.secret = secret_mock
+        provider.uid = provider_uid
+
+        result = get_prowler_provider_kwargs(provider)
+
+        expected_result = {"registry": provider_uid}
+        assert result == expected_result
+
+    def test_get_prowler_provider_kwargs_image_provider_ignores_mutelist(self):
+        """Test that Image provider does NOT receive mutelist_content.
+
+        Image provider uses Trivy's built-in logic, so it should not
+        receive mutelist_content even when a mutelist processor is configured.
+        """
+        provider_uid = "ghcr.io"
+        secret_dict = {"registry_token": "test_token"}
+        secret_mock = MagicMock()
+        secret_mock.secret = secret_dict
+
+        mutelist_processor = MagicMock()
+        mutelist_processor.configuration = {"Mutelist": {"key": "value"}}
+
+        provider = MagicMock()
+        provider.provider = Provider.ProviderChoices.IMAGE.value
+        provider.secret = secret_mock
+        provider.uid = provider_uid
+
+        result = get_prowler_provider_kwargs(provider, mutelist_processor)
+
+        assert "mutelist_content" not in result
+        expected_result = {
+            "registry": provider_uid,
+            "registry_token": "test_token",
+        }
+        assert result == expected_result
+
     def test_get_prowler_provider_kwargs_unsupported_provider(self):
         # Setup
         provider_uid = "provider_uid"

api/src/backend/api/tests/test_views.py

@@ -1189,6 +1189,26 @@ class TestProviderViewSet:
                     "uid": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
                     "alias": "OpenStack Project",
                 },
+                {
+                    "provider": "image",
+                    "uid": "ghcr.io",
+                    "alias": "GitHub Registry",
+                },
+                {
+                    "provider": "image",
+                    "uid": "docker.io",
+                    "alias": "Docker Hub",
+                },
+                {
+                    "provider": "image",
+                    "uid": "123456789012.dkr.ecr.us-east-1.amazonaws.com",
+                    "alias": "ECR",
+                },
+                {
+                    "provider": "image",
+                    "uid": "registry.example.com:5000",
+                    "alias": "Private",
+                },
             ]
         ),
     )
@@ -1202,6 +1222,24 @@ class TestProviderViewSet:
         assert Provider.objects.get().uid == provider_json_payload["uid"]
         assert Provider.objects.get().alias == provider_json_payload["alias"]
 
+    @pytest.mark.parametrize(
+        "uid_input, expected_uid",
+        [
+            ("https://registry.example.com:5000", "registry.example.com:5000"),
+            ("http://registry.local:8080", "registry.local:8080"),
+        ],
+    )
+    def test_providers_create_image_strips_http_prefix(
+        self, authenticated_client, uid_input, expected_uid
+    ):
+        response = authenticated_client.post(
+            reverse("provider-list"),
+            data={"provider": "image", "uid": uid_input, "alias": "test"},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+        assert Provider.objects.get().uid == expected_uid
+
     @pytest.mark.parametrize(
         "provider_json_payload",
         (
@@ -1633,6 +1671,26 @@ class TestProviderViewSet:
                     "min_length",
                     "uid",
                 ),
+                # Image UID validation - too short (below min_length)
+                (
+                    {
+                        "provider": "image",
+                        "uid": "ab",
+                        "alias": "test",
+                    },
+                    "min_length",
+                    "uid",
+                ),
+                # Image UID validation - invalid characters (space)
+                (
+                    {
+                        "provider": "image",
+                        "uid": "not valid!",
+                        "alias": "test",
+                    },
+                    "image-uid",
+                    "uid",
+                ),
             ]
         ),
     )
@@ -1788,6 +1846,38 @@ class TestProviderViewSet:
         assert "Content-Location" in response.headers
         assert response.headers["Content-Location"] == f"/api/v1/tasks/{task_mock.id}"
 
+    @patch("api.v1.views.Task.objects.get")
+    @patch("api.v1.views.check_provider_connection_task.delay")
+    def test_providers_connection_image(
+        self,
+        mock_provider_connection,
+        mock_task_get,
+        authenticated_client,
+        providers_fixture,
+        tasks_fixture,
+    ):
+        prowler_task = tasks_fixture[0]
+        task_mock = Mock()
+        task_mock.id = prowler_task.id
+        task_mock.status = "PENDING"
+        mock_provider_connection.return_value = task_mock
+        mock_task_get.return_value = prowler_task
+
+        image_provider = providers_fixture[11]
+        assert image_provider.provider == "image"
+        assert image_provider.connected is None
+        assert image_provider.connection_last_checked_at is None
+
+        response = authenticated_client.post(
+            reverse("provider-connection", kwargs={"pk": image_provider.id})
+        )
+        assert response.status_code == status.HTTP_202_ACCEPTED
+        mock_provider_connection.assert_called_once_with(
+            provider_id=str(image_provider.id), tenant_id=ANY
+        )
+        assert "Content-Location" in response.headers
+        assert response.headers["Content-Location"] == f"/api/v1/tasks/{task_mock.id}"
+
     def test_providers_connection_invalid_provider(
         self, authenticated_client, providers_fixture
     ):
@@ -1810,17 +1900,17 @@ class TestProviderViewSet:
                 ),
                 ("alias", "aws_testing_1", 1),
                 ("alias.icontains", "aws", 2),
-                ("inserted_at", TODAY, 11),
+                ("inserted_at", TODAY, 12),
                 (
                     "inserted_at.gte",
                     "2024-01-01",
-                    11,
+                    12,
                 ),
                 ("inserted_at.lte", "2024-01-01", 0),
                 (
                     "updated_at.gte",
                     "2024-01-01",
-                    11,
+                    12,
                 ),
                 ("updated_at.lte", "2024-01-01", 0),
             ]
@@ -2436,6 +2526,39 @@ class TestProviderSecretViewSet:
                     "clouds_yaml_cloud": "mycloud",
                 },
             ),
+            # Image with Docker login
+            (
+                Provider.ProviderChoices.IMAGE.value,
+                ProviderSecret.TypeChoices.STATIC,
+                {
+                    "registry_username": "user",
+                    "registry_password": "pass",
+                },
+            ),
+            # Image with token
+            (
+                Provider.ProviderChoices.IMAGE.value,
+                ProviderSecret.TypeChoices.STATIC,
+                {
+                    "registry_token": "ghp_abc123",
+                },
+            ),
+            # Image with no auth + filters
+            (
+                Provider.ProviderChoices.IMAGE.value,
+                ProviderSecret.TypeChoices.STATIC,
+                {
+                    "image_filter": "my-app.*",
+                    "tag_filter": "v[0-9]+",
+                    "max_images": 50,
+                },
+            ),
+            # Image with empty secret (public registry)
+            (
+                Provider.ProviderChoices.IMAGE.value,
+                ProviderSecret.TypeChoices.STATIC,
+                {},
+            ),
         ],
     )
     def test_provider_secrets_create_valid(

api/src/backend/api/utils.py

@@ -28,6 +28,7 @@ if TYPE_CHECKING:
     from prowler.providers.gcp.gcp_provider import GcpProvider
     from prowler.providers.github.github_provider import GithubProvider
     from prowler.providers.iac.iac_provider import IacProvider
+    from prowler.providers.image.image_provider import ImageProvider
     from prowler.providers.kubernetes.kubernetes_provider import KubernetesProvider
     from prowler.providers.m365.m365_provider import M365Provider
     from prowler.providers.mongodbatlas.mongodbatlas_provider import (
@@ -83,6 +84,7 @@ def return_prowler_provider(
     | GcpProvider
     | GithubProvider
     | IacProvider
+    | ImageProvider
     | KubernetesProvider
     | M365Provider
     | MongodbatlasProvider
@@ -95,7 +97,7 @@ def return_prowler_provider(
         provider (Provider): The provider object containing the provider type and associated secrets.
 
     Returns:
-        AlibabacloudProvider | AwsProvider | AzureProvider | CloudflareProvider | GcpProvider | GithubProvider | IacProvider | KubernetesProvider | M365Provider | MongodbatlasProvider | OpenstackProvider | OraclecloudProvider: The corresponding provider class.
+        AlibabacloudProvider | AwsProvider | AzureProvider | CloudflareProvider | GcpProvider | GithubProvider | IacProvider | ImageProvider | KubernetesProvider | M365Provider | MongodbatlasProvider | OpenstackProvider | OraclecloudProvider: The corresponding provider class.
 
     Raises:
         ValueError: If the provider type specified in `provider.provider` is not supported.
@@ -159,6 +161,10 @@ def return_prowler_provider(
             from prowler.providers.openstack.openstack_provider import OpenstackProvider
 
             prowler_provider = OpenstackProvider
+        case Provider.ProviderChoices.IMAGE.value:
+            from prowler.providers.image.image_provider import ImageProvider
+
+            prowler_provider = ImageProvider
         case _:
             raise ValueError(f"Provider type {provider.provider} not supported")
     return prowler_provider
@@ -221,11 +227,29 @@ def get_prowler_provider_kwargs(
         # clouds.yaml is not feasible because not all auth methods include it and the
         # Keystone API is unavailable on public clouds.
         pass
+    elif provider.provider == Provider.ProviderChoices.IMAGE.value:
+        prowler_provider_kwargs = {"registry": provider.uid}
+        for key in (
+            "registry_username",
+            "registry_password",
+            "registry_token",
+            "image_filter",
+            "tag_filter",
+        ):
+            if key in provider.secret.secret:
+                prowler_provider_kwargs[key] = provider.secret.secret[key]
+        if "max_images" in provider.secret.secret:
+            prowler_provider_kwargs["max_images"] = int(
+                provider.secret.secret["max_images"]
+            )
 
     if mutelist_processor:
         mutelist_content = mutelist_processor.configuration.get("Mutelist", {})
-        # IaC provider doesn't support mutelist (uses Trivy's built-in logic)
-        if mutelist_content and provider.provider != Provider.ProviderChoices.IAC.value:
+        # IaC/Image providers don't support mutelist (use Trivy's built-in logic)
+        if mutelist_content and provider.provider not in (
+            Provider.ProviderChoices.IAC.value,
+            Provider.ProviderChoices.IMAGE.value,
+        ):
             prowler_provider_kwargs["mutelist_content"] = mutelist_content
 
     return prowler_provider_kwargs
@@ -242,6 +266,7 @@ def initialize_prowler_provider(
     | GcpProvider
     | GithubProvider
     | IacProvider
+    | ImageProvider
     | KubernetesProvider
     | M365Provider
     | MongodbatlasProvider
@@ -255,7 +280,7 @@ def initialize_prowler_provider(
         mutelist_processor (Processor): The mutelist processor object containing the mutelist configuration.
 
     Returns:
-        AlibabacloudProvider | AwsProvider | AzureProvider | CloudflareProvider | GcpProvider | GithubProvider | IacProvider | KubernetesProvider | M365Provider | MongodbatlasProvider | OpenstackProvider | OraclecloudProvider: An instance of the corresponding provider class
+        AlibabacloudProvider | AwsProvider | AzureProvider | CloudflareProvider | GcpProvider | GithubProvider | IacProvider | ImageProvider | KubernetesProvider | M365Provider | MongodbatlasProvider | OpenstackProvider | OraclecloudProvider: An instance of the corresponding provider class
             initialized with the provider's secrets.
     """
     prowler_provider = return_prowler_provider(provider)
@@ -297,6 +322,20 @@ def prowler_provider_connection_test(provider: Provider) -> Connection:
             "raise_on_exception": False,
         }
         return prowler_provider.test_connection(**openstack_kwargs)
+    elif provider.provider == Provider.ProviderChoices.IMAGE.value:
+        from prowler.providers.image.lib.registry.factory import create_registry_adapter
+
+        try:
+            adapter = create_registry_adapter(
+                registry_url=provider.uid,
+                username=prowler_provider_kwargs.get("registry_username"),
+                password=prowler_provider_kwargs.get("registry_password"),
+                token=prowler_provider_kwargs.get("registry_token"),
+            )
+            adapter.list_repositories()
+            return Connection(is_connected=True)
+        except Exception as e:
+            return Connection(is_connected=False, error=str(e))
     else:
         return prowler_provider.test_connection(
             **prowler_provider_kwargs,

api/src/backend/api/v1/serializer_utils/providers.py

@@ -388,6 +388,37 @@ from rest_framework_json_api import serializers
                 },
                 "required": ["clouds_yaml_content", "clouds_yaml_cloud"],
             },
+            {
+                "type": "object",
+                "title": "Image Registry Credentials",
+                "properties": {
+                    "registry_username": {
+                        "type": "string",
+                        "description": "Username for Docker login authentication.",
+                    },
+                    "registry_password": {
+                        "type": "string",
+                        "description": "Password for Docker login authentication.",
+                    },
+                    "registry_token": {
+                        "type": "string",
+                        "description": "Bearer token for registry authentication.",
+                    },
+                    "image_filter": {
+                        "type": "string",
+                        "description": "Regex pattern to filter repository names during registry enumeration.",
+                    },
+                    "tag_filter": {
+                        "type": "string",
+                        "description": "Regex pattern to filter image tags during registry enumeration.",
+                    },
+                    "max_images": {
+                        "type": "integer",
+                        "minimum": 0,
+                        "description": "Maximum number of images to scan (0 = unlimited).",
+                    },
+                },
+            },
         ]
     }
 )

api/src/backend/api/v1/serializers.py

@@ -1528,6 +1528,8 @@ class BaseWriteProviderSecretSerializer(BaseWriteSerializer):
                     )
             elif provider_type == Provider.ProviderChoices.OPENSTACK.value:
                 serializer = OpenStackCloudsYamlProviderSecret(data=secret)
+            elif provider_type == Provider.ProviderChoices.IMAGE.value:
+                serializer = ImageProviderSecret(data=secret)
             else:
                 raise serializers.ValidationError(
                     {"provider": f"Provider type not supported {provider_type}"}
@@ -1666,6 +1668,41 @@ class IacProviderSecret(serializers.Serializer):
         resource_name = "provider-secrets"
 
 
+class ImageProviderSecret(serializers.Serializer):
+    registry_username = serializers.CharField(required=False)
+    registry_password = serializers.CharField(required=False)
+    registry_token = serializers.CharField(required=False)
+    image_filter = serializers.CharField(required=False)
+    tag_filter = serializers.CharField(required=False)
+    max_images = serializers.IntegerField(required=False, min_value=0)
+
+    def validate(self, attrs):
+        username = attrs.get("registry_username")
+        password = attrs.get("registry_password")
+        token = attrs.get("registry_token")
+
+        if token and (username or password):
+            raise serializers.ValidationError(
+                {"secret": "Provide either username/password or token, not both."}
+            )
+        if username and not password:
+            raise serializers.ValidationError(
+                {
+                    "secret/registry_password": "Password is required when username is provided."
+                }
+            )
+        if password and not username:
+            raise serializers.ValidationError(
+                {
+                    "secret/registry_username": "Username is required when password is provided."
+                }
+            )
+        return attrs
+
+    class Meta:
+        resource_name = "provider-secrets"
+
+
 class OracleCloudProviderSecret(serializers.Serializer):
     user = serializers.CharField()
     fingerprint = serializers.CharField()

api/src/backend/conftest.py

@@ -543,6 +543,12 @@ def providers_fixture(tenants_fixture):
         alias="openstack_testing",
         tenant_id=tenant.id,
     )
+    provider12 = Provider.objects.create(
+        provider="image",
+        uid="ghcr.io",
+        alias="image_testing",
+        tenant_id=tenant.id,
+    )
 
     return (
         provider1,
@@ -556,6 +562,7 @@ def providers_fixture(tenants_fixture):
         provider9,
         provider10,
         provider11,
+        provider12,
     )
 
 

api/src/backend/tasks/jobs/export.py

@@ -137,6 +137,9 @@ COMPLIANCE_CLASS_MAP = {
         # IaC provider doesn't have specific compliance frameworks yet
         # Trivy handles its own compliance checks
     ],
+    "image": [
+        # Image provider doesn't have compliance frameworks yet
+    ],
     "oraclecloud": [
         (lambda name: name.startswith("cis_"), OracleCloudCIS),
         (lambda name: name.startswith("csa_"), OracleCloudCSA),

api/src/backend/tasks/tasks.py

@@ -11,8 +11,8 @@ from django_celery_beat.models import PeriodicTask
 from tasks.jobs.attack_paths import (
     attack_paths_scan,
     can_provider_run_attack_paths_scan,
-    db_utils as attack_paths_db_utils,
 )
+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):