feat(api): integrate Image provider into API layer

- Add IMAGE to ProviderChoices and validate_image_uid
- Add ImageProviderSecret serializer for registry credentials
- Wire Image provider into connection test and scan initialization
- Rebase migration to 0081 after master merge conflict

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

@@ -5,7 +5,7 @@ import api.db_utils
 
 class Migration(migrations.Migration):
     dependencies = [
-        ("api", "0075_cloudflare_provider"),
+        ("api", "0080_backfill_attack_paths_graph_data_ready"),
     ]
 
     operations = [
@@ -25,6 +25,7 @@ class Migration(migrations.Migration):
                     ("oraclecloud", "Oracle Cloud Infrastructure"),
                     ("alibabacloud", "Alibaba Cloud"),
                     ("cloudflare", "Cloudflare"),
+                    ("openstack", "OpenStack"),
                     ("image", "Image"),
                 ],
                 default="aws",

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,15 @@ class Provider(RowLevelSecurityProtectedModel):
                 pointer="/data/attributes/uid",
             )
 
+    @staticmethod
+    def validate_image_uid(value):
+        if not re.match(r"^[a-zA-Z0-9][a-zA-Z0-9._/:@-]{2,249}$", value):
+            raise ModelValidationError(
+                detail="Image provider ID must be a valid container image reference.",
+                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)

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,6 +227,13 @@ 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:
+        # Image provider uses uid as the image reference; registry credentials
+        # come from the secret (all optional).
+        prowler_provider_kwargs = {
+            "images": [provider.uid],
+            **{k: v for k, v in prowler_provider_kwargs.items() if v},
+        }
 
     if mutelist_processor:
         mutelist_content = mutelist_processor.configuration.get("Mutelist", {})
@@ -242,6 +255,7 @@ def initialize_prowler_provider(
     | GcpProvider
     | GithubProvider
     | IacProvider
+    | ImageProvider
     | KubernetesProvider
     | M365Provider
     | MongodbatlasProvider
@@ -255,7 +269,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 +311,22 @@ 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:
+        image_kwargs = {
+            "image": provider.uid,
+            "raise_on_exception": False,
+        }
+        if prowler_provider_kwargs.get("registry_username"):
+            image_kwargs["registry_username"] = prowler_provider_kwargs[
+                "registry_username"
+            ]
+        if prowler_provider_kwargs.get("registry_password"):
+            image_kwargs["registry_password"] = prowler_provider_kwargs[
+                "registry_password"
+            ]
+        if prowler_provider_kwargs.get("registry_token"):
+            image_kwargs["registry_token"] = prowler_provider_kwargs["registry_token"]
+        return prowler_provider.test_connection(**image_kwargs)
     else:
         return prowler_provider.test_connection(
             **prowler_provider_kwargs,

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}"}
@@ -1702,6 +1704,15 @@ class OpenStackCloudsYamlProviderSecret(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)
+
+    class Meta:
+        resource_name = "provider-secrets"
+
+
 class AlibabaCloudProviderSecret(serializers.Serializer):
     access_key_id = serializers.CharField()
     access_key_secret = serializers.CharField()