feat(api): add custom check available method under model

api/src/backend/api/exceptions.py

@@ -70,6 +70,25 @@ class ProviderDeletedException(Exception):
     """Raised when a provider has been deleted during scan/task execution."""
 
 
+class ProviderNotAvailableError(APIException):
+    """Raised when attempting to perform actions on an unavailable provider."""
+
+    status_code = status.HTTP_400_BAD_REQUEST
+    default_detail = (
+        "Cannot perform this action on an unavailable provider. "
+        "The provider no longer exists in the cloud environment."
+    )
+    default_code = "provider_unavailable"
+
+    def __init__(self, detail=None):
+        error_detail = {
+            "detail": detail or self.default_detail,
+            "status": str(self.status_code),
+            "code": self.default_code,
+        }
+        super().__init__(detail=[error_detail])
+
+
 def custom_exception_handler(exc, context):
     if isinstance(exc, django_validation_error):
         if hasattr(exc, "error_dict"):

api/src/backend/api/models.py

@@ -48,7 +48,7 @@ from api.db_utils import (
     generate_random_token,
     one_week_from_now,
 )
-from api.exceptions import ModelValidationError
+from api.exceptions import ModelValidationError, ProviderNotAvailableError
 from api.rls import (
     BaseSecurityConstraint,
     RowLevelSecurityConstraint,
@@ -434,6 +434,16 @@ class Provider(RowLevelSecurityProtectedModel):
         self.full_clean()
         super().save(*args, **kwargs)
 
+    def check_available(self):
+        """
+        Check if the provider is available.
+
+        Raises:
+            ProviderNotAvailableError: If the provider is not available.
+        """
+        if not self.available:
+            raise ProviderNotAvailableError()
+
     class Meta(RowLevelSecurityProtectedModel.Meta):
         db_table = "providers"
 

api/src/backend/api/specs/v1.yaml

@@ -15907,10 +15907,6 @@ components:
                     'Production AWS Account', 'Dev Environment'
                   maxLength: 100
                   minLength: 3
-                available:
-                  type: boolean
-                  description: Whether the provider account still exists. Set to False
-                    if the provider is ephemeral and no longer exists.
       required:
       - data
     PatchedRoleProviderGroupRelationshipRequest:

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

@@ -2914,7 +2914,6 @@ class TestScanViewSet:
                 "type": "scans",
                 "attributes": {
                     "name": "Test Scan",
-                    "trigger": Scan.TriggerChoices.MANUAL,
                 },
                 "relationships": {
                     "provider": {"data": {"type": "providers", "id": str(provider1.id)}}
@@ -2928,7 +2927,7 @@ class TestScanViewSet:
             content_type=API_JSON_CONTENT_TYPE,
         )
         assert response.status_code == status.HTTP_400_BAD_REQUEST
-        assert "unavailable provider" in response.json()["errors"][0]["detail"].lower()
+        assert response.json()["errors"][0]["code"] == "provider_unavailable"
 
     def test_scans_partial_update(self, authenticated_client, scans_fixture):
         scan1, *_ = scans_fixture
@@ -8209,6 +8208,23 @@ class TestScheduleViewSet:
         )
         assert response.status_code == status.HTTP_409_CONFLICT
 
+    def test_schedule_daily_unavailable_provider(
+        self, authenticated_client, providers_fixture
+    ):
+        """Test that creating a schedule for an unavailable provider returns 400."""
+        provider, *_ = providers_fixture
+        provider.available = False
+        provider.save()
+
+        json_payload = {
+            "provider_id": str(provider.id),
+        }
+        response = authenticated_client.post(
+            reverse("schedule-daily"), data=json_payload, format="json"
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.json()["errors"][0]["code"] == "provider_unavailable"
+
 
 @pytest.mark.django_db
 class TestIntegrationViewSet:

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

@@ -964,23 +964,19 @@ class ProviderCreateSerializer(RLSSerializer, BaseWriteSerializer):
 class ProviderUpdateSerializer(BaseWriteSerializer):
     """
     Serializer for updating the Provider model.
-    Only allows "alias", "available" and "scanner_args" fields to be updated.
+    Only allows "alias" and "scanner_args" fields to be updated.
     """
 
     class Meta:
         model = Provider
         fields = [
             "alias",
-            "available",
             # "scanner_args"
         ]
         extra_kwargs = {
             "alias": {
                 "help_text": "Human readable name to identify the provider, e.g. 'Production AWS Account', 'Dev Environment'",
-            },
-            "available": {
-                "help_text": "Whether the provider account still exists. Set to False if the provider is ephemeral and no longer exists.",
-            },
+            }
         }
 
 
@@ -1066,14 +1062,6 @@ class ScanCreateSerializer(RLSSerializer, BaseWriteSerializer):
             "name",
         ]
 
-    def validate_provider(self, provider):
-        if not provider.available:
-            raise serializers.ValidationError(
-                "Cannot create scan for unavailable provider. "
-                "The provider no longer exists."
-            )
-        return provider
-
     def create(self, validated_data):
         # provider = validated_data.get("provider")
 
@@ -2364,6 +2352,11 @@ class ScheduleDailyCreateSerializer(BaseSerializerV1):
     class JSONAPIMeta:
         resource_name = "daily-schedules"
 
+    def validate_provider_id(self, provider_id):
+        if not Provider.objects.filter(pk=provider_id).exists():
+            raise serializers.ValidationError("Provider not found.")
+        return provider_id
+
     # TODO: DRY this when we have more time
     def validate(self, data):
         if hasattr(self, "initial_data"):

api/src/backend/api/v1/views.py

@@ -1552,21 +1552,7 @@ class ProviderViewSet(DisablePaginationMixin, BaseRLSViewSet):
     @action(detail=True, methods=["post"], url_name="connection")
     def connection(self, request, pk=None):
         provider = get_object_or_404(Provider, pk=pk)
-
-        if not provider.available:
-            return Response(
-                data={
-                    "errors": [
-                        {
-                            "detail": "Cannot check connection for unavailable provider. "
-                            "The provider no longer exists.",
-                            "status": "400",
-                            "code": "provider_unavailable",
-                        }
-                    ]
-                },
-                status=status.HTTP_400_BAD_REQUEST,
-            )
+        provider.check_available()
 
         with transaction.atomic():
             task = check_provider_connection_task.delay(
@@ -2158,6 +2144,12 @@ class ScanViewSet(BaseRLSViewSet):
     def create(self, request, *args, **kwargs):
         input_serializer = self.get_serializer(data=request.data)
         input_serializer.is_valid(raise_exception=True)
+
+        # Check provider availability before creating scan
+        provider = input_serializer.validated_data.get("provider")
+        if provider:
+            provider.check_available()
+
         with transaction.atomic():
             scan = input_serializer.save()
         with transaction.atomic():
@@ -5146,6 +5138,8 @@ class ScheduleViewSet(BaseRLSViewSet):
         provider_id = serializer.validated_data["provider_id"]
 
         provider_instance = get_object_or_404(Provider, pk=provider_id)
+        provider_instance.check_available()
+
         with transaction.atomic():
             task = schedule_provider_scan(provider_instance)
 

api/src/backend/tasks/tests/test_scan.py

@@ -260,7 +260,7 @@ class TestPerformScan:
         assert provider.connected is False
         assert isinstance(provider.connection_last_checked_at, datetime)
 
-    @patch("api.db_utils.rls_transaction")
+    @patch("tasks.jobs.scan.rls_transaction")
     def test_perform_prowler_scan_unavailable_provider(
         self,
         mock_rls_transaction,