Prowler Api

1# When to use admin (bypasses RLS)
2from api.db_router import MainRouter
3User.objects.using(MainRouter.admin_db).get(id=user_id)  # Auth lookups
4
5# Standard queries use default (RLS enforced)
6Provider.objects.filter(connected=True)  # Requires rls_transaction context

 1from api.rls import RowLevelSecurityProtectedModel, RowLevelSecurityConstraint
 2
 3class MyModel(RowLevelSecurityProtectedModel):
 4    # tenant FK inherited from parent
 5    id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
 6    name = models.CharField(max_length=255)
 7    inserted_at = models.DateTimeField(auto_now_add=True, editable=False)
 8    updated_at = models.DateTimeField(auto_now=True, editable=False)
 9
10    class Meta(RowLevelSecurityProtectedModel.Meta):
11        db_table = "my_models"
12        constraints = [
13            RowLevelSecurityConstraint(
14                field="tenant_id",
15                name="rls_on_%(class)s",
16                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
17            ),
18        ]
19
20    class JSONAPIMeta:
21        resource_name = "my-models"

 1class Resource(RowLevelSecurityProtectedModel):
 2    tags = models.ManyToManyField(
 3        ResourceTag,
 4        through="ResourceTagMapping",  # REQUIRED for RLS
 5    )
 6
 7class ResourceTagMapping(RowLevelSecurityProtectedModel):
 8    # Through model MUST have tenant_id for RLS
 9    resource = models.ForeignKey(Resource, on_delete=models.CASCADE)
10    tag = models.ForeignKey(ResourceTag, on_delete=models.CASCADE)
11
12    class Meta:
13        constraints = [
14            RowLevelSecurityConstraint(
15                field="tenant_id",
16                name="rls_on_%(class)s",
17                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
18            ),
19        ]

 1@action(detail=True, methods=["post"], url_name="connection")
 2def connection(self, request, pk=None):
 3    with transaction.atomic():
 4        task = check_provider_connection_task.delay(
 5            provider_id=pk, tenant_id=self.request.tenant_id
 6        )
 7    prowler_task = Task.objects.get(id=task.id)
 8    serializer = TaskSerializer(prowler_task)
 9    return Response(
10        data=serializer.data,
11        status=status.HTTP_202_ACCEPTED,
12        headers={"Content-Location": reverse("task-detail", kwargs={"pk": prowler_task.id})}
13    )

1def get_queryset(self):
2    user_role = get_role(self.request.user)
3    if user_role.unlimited_visibility:
4        return Model.objects.filter(tenant_id=self.request.tenant_id)
5    else:
6        # Filter by provider_groups assigned to role
7        return Model.objects.filter(provider__in=get_providers(user_role))

1from uuid6 import uuid7
2from api.uuid_utils import uuid7_start, uuid7_end, datetime_to_uuid7
3
4# Partition-aware filtering
5start = uuid7_start(datetime_to_uuid7(date_from))
6end = uuid7_end(datetime_to_uuid7(date_to), settings.FINDINGS_TABLE_PARTITION_MONTHS)
7queryset.filter(id__gte=start, id__lt=end)

 1from api.db_utils import batch_delete, create_objects_in_batches, update_objects_in_batches
 2
 3# Delete in batches (RLS-aware)
 4batch_delete(tenant_id, queryset, batch_size=1000)
 5
 6# Bulk create with RLS
 7create_objects_in_batches(tenant_id, Finding, objects, batch_size=500)
 8
 9# Bulk update with RLS
10update_objects_in_batches(tenant_id, Finding, objects, fields=["status"], batch_size=500)

 1# Safe task queuing - task only enqueued after transaction commits
 2with transaction.atomic():
 3    provider = Provider.objects.create(**data)
 4    transaction.on_commit(
 5        lambda: verify_provider_connection.delay(
 6            tenant_id=str(request.tenant_id),
 7            provider_id=str(provider.id)
 8        )
 9    )
10
11# Modern retry pattern
12@shared_task(
13    base=RLSTask,
14    bind=True,
15    autoretry_for=(ConnectionError, TimeoutError, OperationalError),
16    retry_backoff=True,
17    retry_backoff_max=600,
18    retry_jitter=True,
19    max_retries=5,
20    soft_time_limit=300,
21    time_limit=360,
22)
23@set_tenant
24def sync_provider_data(self, tenant_id, provider_id):
25    with rls_transaction(tenant_id):
26        # ... task logic
27        pass
28
29# Idempotent task - safe to retry
30@shared_task(base=RLSTask, acks_late=True)
31@set_tenant
32def process_finding(tenant_id, finding_uid, data):
33    with rls_transaction(tenant_id):
34        Finding.objects.update_or_create(uid=finding_uid, defaults=data)

1cd api && poetry run python src/backend/manage.py check --deploy

 1# Development
 2cd api && poetry run python src/backend/manage.py runserver
 3cd api && poetry run python src/backend/manage.py shell
 4
 5# Celery
 6cd api && poetry run celery -A config.celery worker -l info -Q scans,overview
 7cd api && poetry run celery -A config.celery beat -l info
 8
 9# Testing
10cd api && poetry run pytest -x --tb=short
11
12# Production checks
13cd api && poetry run python src/backend/manage.py check --deploy