feat(Provider, Scan): PRWLR-5405 Add daily schedule scan for added Providers (#87)

* feat: PRWLR-5405 add django-celery-beat dep

* fix(Scan): PRWLR-5405 fix scan status update in real-time

* feat(Celery-beat): PRWLR-5405 add django-celery-beat resources

* feat(Provider,Beat): PRWLR-5405 add daily scheduled scan for each new provider

* chore(Scan): PRWLR-5405 change default sort parameter

* fix(Migrations): PRWLR-5405 fix meta fields

* test(Tenants): PRWLR-5405 fix integration tests

* chore: fix typo on .env.example

.env.example

@@ -19,7 +19,7 @@ DJANGO_REFRESH_TOKEN_LIFETIME=1440
 DJANGO_CACHE_MAX_AGE=3600
 DJANGO_STALE_WHILE_REVALIDATE=60
 DJANGO_SECRETS_ENCRYPTION_KEY=""
-# Decide whether to allow Djanto manage database table partitions
+# Decide whether to allow Django manage database table partitions
 DJANGO_MANAGE_DB_PARTITIONS=[True|False]
 
 # PostgreSQL settings

docker-compose.yml

@@ -87,7 +87,7 @@ services:
       dockerfile: Dockerfile
     image: prowler-worker
     environment:
-      - DJANGO_SETTINGS_MODULE=config.django.production
+      - DJANGO_SETTINGS_MODULE=${DJANGO_SETTINGS_MODULE:-config.django.production}
     env_file:
       - path: ./.env
         required: false
@@ -102,3 +102,24 @@ services:
     entrypoint:
       - "../docker-entrypoint.sh"
       - "worker"
+
+  worker-beat:
+    build:
+      dockerfile: Dockerfile
+    image: prowler-worker
+    environment:
+      - DJANGO_SETTINGS_MODULE=${DJANGO_SETTINGS_MODULE:-config.django.production}
+    env_file:
+      - path: ./.env
+        required: false
+    profiles:
+      - dev
+      - prod
+    depends_on:
+      valkey:
+        condition: service_healthy
+      postgres:
+        condition: service_healthy
+    entrypoint:
+      - "../docker-entrypoint.sh"
+      - "beat"

docker-entrypoint.sh

@@ -31,6 +31,12 @@ start_worker() {
   poetry run python -m celery -A config.celery worker -l "${DJANGO_LOGGING_LEVEL:-info}" -Q celery,scans -E
 }
 
+start_worker_beat() {
+  echo "Starting the worker-beat..."
+  sleep 15
+  poetry run python -m celery -A config.celery beat -l "${DJANGO_LOGGING_LEVEL:-info}" --scheduler django_celery_beat.schedulers:DatabaseScheduler
+}
+
 manage_db_partitions() {
   if [ "${DJANGO_MANAGE_DB_PARTITIONS}" = "True" ]; then
     echo "Managing DB partitions..."
@@ -55,8 +61,11 @@ case "$1" in
   worker)
     start_worker
     ;;
+  beat)
+    start_worker_beat
+    ;;
   *)
-    echo "Usage: $0 {dev|prod|worker}"
+    echo "Usage: $0 {dev|prod|worker|beat}"
     exit 1
     ;;
 esac

poetry.lock

@@ -1117,6 +1117,20 @@ files = [
 [package.extras]
 toml = ["tomli"]
 
+[[package]]
+name = "cron-descriptor"
+version = "1.4.5"
+description = "A Python library that converts cron expressions into human readable strings."
+optional = false
+python-versions = "*"
+files = [
+    {file = "cron_descriptor-1.4.5-py3-none-any.whl", hash = "sha256:736b3ae9d1a99bc3dbfc5b55b5e6e7c12031e7ba5de716625772f8b02dcd6013"},
+    {file = "cron_descriptor-1.4.5.tar.gz", hash = "sha256:f51ce4ffc1d1f2816939add8524f206c376a42c87a5fca3091ce26725b3b1bca"},
+]
+
+[package.extras]
+dev = ["polib"]
+
 [[package]]
 name = "cryptography"
 version = "43.0.1"
@@ -1355,6 +1369,25 @@ tzdata = {version = "*", markers = "sys_platform == \"win32\""}
 argon2 = ["argon2-cffi (>=19.1.0)"]
 bcrypt = ["bcrypt"]
 
+[[package]]
+name = "django-celery-beat"
+version = "2.7.0"
+description = "Database-backed Periodic Tasks."
+optional = false
+python-versions = ">=3.8"
+files = [
+    {file = "django_celery_beat-2.7.0-py3-none-any.whl", hash = "sha256:851c680d8fbf608ca5fecd5836622beea89fa017bc2b3f94a5b8c648c32d84b1"},
+    {file = "django_celery_beat-2.7.0.tar.gz", hash = "sha256:8482034925e09b698c05ad61c36ed2a8dbc436724a3fe119215193a4ca6dc967"},
+]
+
+[package.dependencies]
+celery = ">=5.2.3,<6.0"
+cron-descriptor = ">=1.2.32"
+Django = ">=2.2,<5.2"
+django-timezone-field = ">=5.0"
+python-crontab = ">=2.3.4"
+tzdata = "*"
+
 [[package]]
 name = "django-celery-results"
 version = "2.5.1"
@@ -1450,6 +1483,20 @@ docs = ["Sphinx (==2.2.0)", "docutils (<0.18)", "sphinx-rtd-theme (==0.4.3)"]
 publish = ["build (==0.7.0)", "twine (==3.7.1)"]
 test = ["coveralls (==3.3.0)", "dj-database-url (==0.5.0)", "freezegun (==1.1.0)", "psycopg2 (>=2.8.4,<3.0.0)", "pytest (==6.2.5)", "pytest-benchmark (==3.4.1)", "pytest-cov (==3.0.0)", "pytest-django (==4.4.0)", "pytest-freezegun (==0.4.2)", "pytest-lazy-fixture (==0.6.3)", "snapshottest (==0.6.0)", "tox (==3.24.4)"]
 
+[[package]]
+name = "django-timezone-field"
+version = "7.0"
+description = "A Django app providing DB, form, and REST framework fields for zoneinfo and pytz timezone objects."
+optional = false
+python-versions = "<4.0,>=3.8"
+files = [
+    {file = "django_timezone_field-7.0-py3-none-any.whl", hash = "sha256:3232e7ecde66ba4464abb6f9e6b8cc739b914efb9b29dc2cf2eee451f7cc2acb"},
+    {file = "django_timezone_field-7.0.tar.gz", hash = "sha256:aa6f4965838484317b7f08d22c0d91a53d64e7bbbd34264468ae83d4023898a7"},
+]
+
+[package.dependencies]
+Django = ">=3.2,<6.0"
+
 [[package]]
 name = "djangorestframework"
 version = "3.15.2"
@@ -3783,6 +3830,24 @@ psutil = ["psutil (>=3.0)"]
 setproctitle = ["setproctitle"]
 testing = ["filelock"]
 
+[[package]]
+name = "python-crontab"
+version = "3.2.0"
+description = "Python Crontab API"
+optional = false
+python-versions = "*"
+files = [
+    {file = "python_crontab-3.2.0-py3-none-any.whl", hash = "sha256:82cb9b6a312d41ff66fd3caf3eed7115c28c195bfb50711bc2b4b9592feb9fe5"},
+    {file = "python_crontab-3.2.0.tar.gz", hash = "sha256:40067d1dd39ade3460b2ad8557c7651514cd3851deffff61c5c60e1227c5c36b"},
+]
+
+[package.dependencies]
+python-dateutil = "*"
+
+[package.extras]
+cron-description = ["cron-descriptor"]
+cron-schedule = ["croniter"]
+
 [[package]]
 name = "python-dateutil"
 version = "2.9.0.post0"
@@ -4895,4 +4960,4 @@ type = ["pytest-mypy"]
 [metadata]
 lock-version = "2.0"
 python-versions = ">=3.11,<3.13"
-content-hash = "58f90b3d204474e915f555faf7deb4d7a9da7b6f6088e7c744e7e0c3784667c0"
+content-hash = "b9ad538d2d86a404dd212731bd1eb6555b6a72a232870f5da24da6b6bc7aa040"

pyproject.toml

@@ -13,6 +13,7 @@ version = "1.0.0"
 [tool.poetry.dependencies]
 celery = {extras = ["pytest"], version = "^5.4.0"}
 django = "5.1.1"
+django-celery-beat = "^2.7.0"
 django-celery-results = "^2.5.1"
 django-cors-headers = "4.4.0"
 django-environ = "0.11.2"

src/backend/api/models.py

@@ -681,6 +681,8 @@ class ResourceFindingMapping(PostgresPartitionedModel, RowLevelSecurityProtected
 
     class Meta(RowLevelSecurityProtectedModel.Meta):
         db_table = "resource_finding_mappings"
+        base_manager_name = "objects"
+        abstract = False
 
         # django will automatically create indexes for:
         #   - resource_id

src/backend/api/signals.py

@@ -1,7 +1,11 @@
 from celery import states
 from celery.signals import before_task_publish
+from django.db.models.signals import post_delete
+from django.dispatch import receiver
+from django_celery_beat.models import PeriodicTask
 from django_celery_results.backends.database import DatabaseBackend
 
+from api.models import Provider
 from config.celery import celery_app
 
 
@@ -22,3 +26,10 @@ def create_task_result_on_publish(sender=None, headers=None, **kwargs):  # noqa:
 before_task_publish.connect(
     create_task_result_on_publish, dispatch_uid="create_task_result_on_publish"
 )
+
+
+@receiver(post_delete, sender=Provider)
+def delete_provider_scan_task(sender, instance, **kwargs):  # noqa: F841
+    # Delete the associated periodic task when the provider is deleted
+    task_name = f"scan-perform-scheduled-{instance.id}"
+    PeriodicTask.objects.filter(name=task_name).delete()

src/backend/api/tests/integration/test_tenants.py

@@ -1,12 +1,18 @@
+from unittest.mock import patch
+
 import pytest
 from django.urls import reverse
 
 from conftest import TEST_USER, TEST_PASSWORD, get_api_tokens, get_authorization_header
 
 
+@patch("tasks.beat.schedule_provider_scan")
 @pytest.mark.django_db
 def test_check_resources_between_different_tenants(
-    enforce_test_user_db_connection, authenticated_api_client, tenants_fixture
+    schedule_mock,
+    enforce_test_user_db_connection,
+    authenticated_api_client,
+    tenants_fixture,
 ):
     client = authenticated_api_client
 

src/backend/api/v1/views.py

@@ -92,6 +92,7 @@ from api.v1.serializers import (
     ComplianceOverviewSerializer,
     ComplianceOverviewFullSerializer,
 )
+from tasks.beat import schedule_provider_scan
 from tasks.tasks import (
     check_provider_connection_task,
     delete_provider_task,
@@ -707,6 +708,14 @@ class ProviderViewSet(BaseRLSViewSet):
             },
         )
 
+    def create(self, request, *args, **kwargs):
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        provider = serializer.save()
+        # Schedule a daily scan for the new provider
+        schedule_provider_scan(provider)
+        return Response(data=serializer.data, status=status.HTTP_201_CREATED)
+
 
 @extend_schema_view(
     list=extend_schema(
@@ -745,7 +754,7 @@ class ScanViewSet(BaseRLSViewSet):
     serializer_class = ScanSerializer
     http_method_names = ["get", "post", "patch"]
     filterset_class = ScanFilter
-    ordering = ["-id"]
+    ordering = ["-inserted_at"]
     ordering_fields = [
         "name",
         "trigger",

src/backend/config/celery.py

@@ -34,9 +34,13 @@ class RLSTask(Task):
             **options,
         )
         task_result_instance = TaskResult.objects.get(task_id=result.task_id)
-        APITask.objects.create(
-            id=task_result_instance.task_id,
-            tenant_id=kwargs.get("tenant_id"),
-            task_runner_task=task_result_instance,
-        )
+        from api.db_utils import tenant_transaction
+
+        tenant_id = kwargs.get("tenant_id")
+        with tenant_transaction(tenant_id):
+            APITask.objects.create(
+                id=task_result_instance.task_id,
+                tenant_id=tenant_id,
+                task_runner_task=task_result_instance,
+            )
         return result

src/backend/config/django/base.py

@@ -27,6 +27,7 @@ INSTALLED_APPS = [
     "django_guid",
     "rest_framework_json_api",
     "django_celery_results",
+    "django_celery_beat",
     "rest_framework_simplejwt.token_blacklist",
 ]
 

src/backend/tasks/beat.py

@@ -0,0 +1,31 @@
+import json
+
+from django.utils import timezone
+from django_celery_beat.models import PeriodicTask, IntervalSchedule
+
+from api.models import Provider
+
+
+def schedule_provider_scan(provider_instance: Provider):
+    schedule, _ = IntervalSchedule.objects.get_or_create(
+        every=24,
+        period=IntervalSchedule.HOURS,
+    )
+
+    # Create a unique name for the periodic task
+    task_name = f"scan-perform-scheduled-{provider_instance.id}"
+
+    # Schedule the task
+    PeriodicTask.objects.create(
+        interval=schedule,
+        name=task_name,
+        task="scan-perform-scheduled",
+        kwargs=json.dumps(
+            {
+                "tenant_id": str(provider_instance.tenant_id),
+                "provider_id": str(provider_instance.id),
+            }
+        ),
+        start_time=provider_instance.inserted_at + timezone.timedelta(hours=24),
+        one_off=False,
+    )

src/backend/tasks/jobs/scan.py

@@ -125,7 +125,8 @@ def perform_prowler_scan(
         scan_instance.started_at = datetime.now(tz=timezone.utc)
         scan_instance.save()
 
-        try:
+    try:
+        with tenant_transaction(tenant_id):
             try:
                 prowler_provider = initialize_prowler_provider(provider_instance)
                 provider_instance.connected = True
@@ -140,18 +141,15 @@ def perform_prowler_scan(
                 )
                 provider_instance.save()
 
-            generate_compliance = (
-                provider_instance.provider != Provider.ProviderChoices.GCP
-            )
-            prowler_scan = ProwlerScan(
-                provider=prowler_provider, checks=checks_to_execute
-            )
+        generate_compliance = provider_instance.provider != Provider.ProviderChoices.GCP
+        prowler_scan = ProwlerScan(provider=prowler_provider, checks=checks_to_execute)
 
-            resource_cache = {}
-            tag_cache = {}
-            last_status_cache = {}
+        resource_cache = {}
+        tag_cache = {}
+        last_status_cache = {}
 
-            for progress, findings in prowler_scan.scan():
+        for progress, findings in prowler_scan.scan():
+            with tenant_transaction(tenant_id):
                 for finding in findings:
                     # Process resource
                     resource_uid = finding.resource_uid
@@ -165,7 +163,7 @@ def perform_prowler_scan(
                                 "region": finding.region,
                                 "service": finding.service_name,
                                 "type": finding.resource_type,
-                                "name": finding.resource_name or "",
+                                "name": finding.resource_name,
                             },
                         )
                         resource_cache[resource_uid] = resource_instance
@@ -251,18 +249,20 @@ def perform_prowler_scan(
                         continue
                     region_dict[finding.check_id] = finding.status.value
 
-                # Update scan progress
+            # Update scan progress
+            with tenant_transaction(tenant_id):
                 scan_instance.progress = progress
                 scan_instance.save()
 
-            scan_instance.state = StateChoices.COMPLETED
+        scan_instance.state = StateChoices.COMPLETED
 
-        except Exception as e:
-            logger.error(f"Error performing scan {scan_id}: {e}")
-            exception = e
-            scan_instance.state = StateChoices.FAILED
+    except Exception as e:
+        logger.error(f"Error performing scan {scan_id}: {e}")
+        exception = e
+        scan_instance.state = StateChoices.FAILED
 
-        finally:
+    finally:
+        with tenant_transaction(tenant_id):
             scan_instance.duration = time.time() - start_time
             scan_instance.completed_at = datetime.now(tz=timezone.utc)
             scan_instance.unique_resource_count = len(unique_resources)

src/backend/tasks/tasks.py

@@ -1,7 +1,8 @@
 from celery import shared_task
 
+from api.db_utils import tenant_transaction
 from api.decorators import set_tenant
-from api.models import Provider
+from api.models import Provider, Scan
 from config.celery import RLSTask
 from tasks.jobs.connection import check_provider_connection
 from tasks.jobs.deletion import delete_instance
@@ -69,3 +70,43 @@ def perform_scan_task(
         provider_id=provider_id,
         checks_to_execute=checks_to_execute,
     )
+
+
+@shared_task(base=RLSTask, bind=True, name="scan-perform-scheduled", queue="scans")
+def perform_scheduled_scan_task(self, tenant_id: str, provider_id: str):
+    """
+    Task to perform a scheduled Prowler scan on a given provider.
+
+    This task creates and executes a Prowler scan for the provider identified by `provider_id`
+    under the tenant identified by `tenant_id`. It initiates a new scan instance with the task ID
+    for tracking purposes. This task is intended to be run on a schedule (e.g., daily) to
+    automatically perform scans without manual intervention.
+
+    Args:
+        self: The task instance (automatically passed when bind=True).
+        tenant_id (str): The tenant ID under which the scan is being performed.
+        provider_id (str): The primary key of the Provider instance to scan.
+
+    Returns:
+        dict: The result of the scan execution, typically including the status and results
+        of the performed checks.
+
+    """
+    task_id = self.request.id
+
+    with tenant_transaction(tenant_id):
+        provider_instance = Provider.objects.get(pk=provider_id)
+
+        scan_instance = Scan.objects.create(
+            tenant_id=tenant_id,
+            name="Daily scheduled scan",
+            provider=provider_instance,
+            trigger=Scan.TriggerChoices.SCHEDULED,
+            task_id=task_id,
+        )
+
+    return perform_prowler_scan(
+        tenant_id=tenant_id,
+        scan_id=str(scan_instance.id),
+        provider_id=provider_id,
+    )