prowler-api

prowler-cloud

Prowler API patterns: RLS, RBAC, providers, Celery tasks. Trigger: When working in api/ on models/serializers/viewsets/filters/tasks involving tenant isolation (RLS), RBAC, or provider lifecycle.

Quick Install

bunx add-skill prowler-cloud/prowler -s prowler-api

Read,Edit,Write,Glob,Grep,Bash,WebFetch,WebSearch,Taskawsazurecis-benchmarkcloudcloudsecuritycompliance

Instructions

When to Use

Use this skill for Prowler-specific patterns:

Row-Level Security (RLS) / tenant isolation
RBAC permissions and role checks
Provider lifecycle and validation
Celery tasks with tenant context
Multi-database architecture (4-database setup)

For generic DRF patterns (ViewSets, Serializers, Filters, JSON:API), use django-drf skill.

Critical Rules

ALWAYS use rls_transaction(tenant_id) when querying outside ViewSet context
ALWAYS use get_role() before checking permissions (returns FIRST role only)
ALWAYS use @set_tenant then @handle_provider_deletion decorator order
ALWAYS use explicit through models for M2M relationships (required for RLS)
NEVER access Provider.objects without RLS context in Celery tasks
NEVER bypass RLS by using raw SQL or

Loading…

Related Skills

flow

Use when you need to run Flow type checking, or when seeing Flow type errors in React code.

242,585

verify

Use when you want to validate changes before committing, or when you need to check all React contribution requirements.

242,585

feature-flags

Use when feature flag tests fail, flags need updating, understanding @gate pragmas, debugging channel-specific test failures, or adding new flags to React.

242,585

flags

Use when you need to check feature flag states, compare channels, or debug why a feature behaves differently across release channels.

242,582

connection.cursor()

# When to use admin (bypasses RLS)
from api.db_router import MainRouter
User.objects.using(MainRouter.admin_db).get(id=user_id)  # Auth lookups

# Standard queries use default (RLS enforced)
Provider.objects.filter(connected=True)  # Requires rls_transaction context

Request → Authentication → BaseRLSViewSet.initial()
                                    │
                                    ├─ Extract tenant_id from JWT
                                    ├─ SET api.tenant_id = 'uuid' (PostgreSQL)
                                    └─ All queries now tenant-scoped

#	Pattern	Reference	Key Points
1	RLS Models	`api/rls.py`	Inherit `RowLevelSecurityProtectedModel`, add constraint
2	RLS Transactions	`api/db_utils.py`	Use `rls_transaction(tenant_id)` context manager
3	RBAC Permissions	`api/rbac/permissions.py`	`get_role()`, `get_providers()`, `Permissions` enum
4	Provider Validation	`api/models.py`	`validate_<provider>_uid()` methods on `Provider` model
5	Celery Tasks	`tasks/tasks.py`, `api/decorators.py`, `config/celery.py`	Task definitions, decorators (`@set_tenant`, `@handle_provider_deletion`), `RLSTask` base
6	RLS Serializers	`api/v1/serializers.py`	Inherit `RLSSerializer` to auto-inject `tenant_id`
7	Through Models	`api/models.py`	ALL M2M must use explicit through with `tenant_id`

Tenant-scoped data       → RowLevelSecurityProtectedModel
Global/shared data       → models.Model + BaseSecurityConstraint (rare)
Partitioned time-series  → PostgresPartitionedModel + RowLevelSecurityProtectedModel
Soft-deletable           → Add is_deleted + ActiveProviderManager

Normal queries           → Model.objects (excludes deleted)
Include deleted records  → Model.all_objects
Celery task context      → Must use rls_transaction() first

Standard API queries     → default (automatic via ViewSet)
Read-only operations     → replica (automatic for GET in BaseRLSViewSet)
Auth/admin operations    → MainRouter.admin_db
Cross-tenant lookups     → MainRouter.admin_db (use sparingly!)

@shared_task(base=RLSTask, name="...", queue="...")
@set_tenant                    # First: sets tenant context
@handle_provider_deletion      # Second: handles deleted providers
def my_task(tenant_id, provider_id):
    pass

from api.rls import RowLevelSecurityProtectedModel, RowLevelSecurityConstraint

class MyModel(RowLevelSecurityProtectedModel):
    # tenant FK inherited from parent
    id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
    name = models.CharField(max_length=255)
    inserted_at = models.DateTimeField(auto_now_add=True, editable=False)
    updated_at = models.DateTimeField(auto_now=True, editable=False)

    class Meta(RowLevelSecurityProtectedModel.Meta):
        db_table = "my_models"
        constraints = [
            RowLevelSecurityConstraint(
                field="tenant_id",
                name="rls_on_%(class)s",
                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
            ),
        ]

    class JSONAPIMeta:
        resource_name = "my-models"

class Resource(RowLevelSecurityProtectedModel):
    tags = models.ManyToManyField(
        ResourceTag,
        through="ResourceTagMapping",  # REQUIRED for RLS
    )

class ResourceTagMapping(RowLevelSecurityProtectedModel):
    # Through model MUST have tenant_id for RLS
    resource = models.ForeignKey(Resource, on_delete=models.CASCADE)
    tag = models.ForeignKey(ResourceTag, on_delete=models.CASCADE)

    class Meta:
        constraints = [
            RowLevelSecurityConstraint(
                field="tenant_id",
                name="rls_on_%(class)s",
                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
            ),
        ]

@action(detail=True, methods=["post"], url_name="connection")
def connection(self, request, pk=None):
    with transaction.atomic():
        task = check_provider_connection_task.delay(
            provider_id=pk, tenant_id=self.request.tenant_id
        )
    prowler_task = Task.objects.get(id=task.id)
    serializer = TaskSerializer(prowler_task)
    return Response(
        data=serializer.data,
        status=status.HTTP_202_ACCEPTED,
        headers={"Content-Location": reverse("task-detail", kwargs={"pk": prowler_task.id})}
    )

Provider	UID Format	Example
AWS	12 digits	`123456789012`
Azure	UUID v4	`a1b2c3d4-e5f6-...`
GCP	6-30 chars, lowercase, letter start	`my-gcp-project`
M365	Valid domain	`contoso.onmicrosoft.com`
Kubernetes	2-251 chars	`arn:aws:eks:...`
GitHub	1-39 chars	`my-org`
IaC	Git URL	`https://github.com/user/repo.git`
Oracle Cloud	OCID format	`ocid1.tenancy.oc1..`
MongoDB Atlas	24-char hex	`507f1f77bcf86cd799439011`
Alibaba Cloud	16 digits	`1234567890123456`

Permission	Controls
`MANAGE_USERS`	User CRUD, role assignments
`MANAGE_ACCOUNT`	Tenant settings
`MANAGE_BILLING`	Billing/subscription
`MANAGE_PROVIDERS`	Provider CRUD
`MANAGE_INTEGRATIONS`	Integration config
`MANAGE_SCANS`	Scan execution
`UNLIMITED_VISIBILITY`	See all providers (bypasses provider_groups)

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

Queue	Purpose
`scans`	Prowler scan execution
`overview`	Dashboard aggregations (severity, attack surface)
`compliance`	Compliance report generation
`integrations`	External integrations (Jira, S3, Security Hub)
`deletion`	Provider/tenant deletion (async)
`backfill`	Historical data backfill operations
`scan-reports`	Output generation (CSV, JSON, HTML, PDF)

Operation	Key Points
Create schedule	`IntervalSchedule.objects.get_or_create(every=24, period=HOURS)`
Create periodic task	Use task name (not function), `kwargs=json.dumps(...)`
Delete scheduled task	`PeriodicTask.objects.filter(name=...).delete()`
Avoid race conditions	Use `countdown=5` to wait for DB commit

Pattern	Description
`bind=True`	Access `self.request.id`, `self.request.retries`
`get_task_logger(__name__)`	Proper logging in Celery tasks
`SoftTimeLimitExceeded`	Catch to save progress before hard kill
`countdown=30`	Defer execution by N seconds
`eta=datetime(...)`	Execute at specific time

Setting	Value	Purpose
`BROKER_VISIBILITY_TIMEOUT`	`86400` (24h)	Prevent re-queue for long tasks
`CELERY_RESULT_BACKEND`	`django-db`	Store results in PostgreSQL
`CELERY_TASK_TRACK_STARTED`	`True`	Track when tasks start
`soft_time_limit`	Task-specific	Raises `SoftTimeLimitExceeded`
`time_limit`	Task-specific	Hard kill (SIGKILL)

from uuid6 import uuid7
from api.uuid_utils import uuid7_start, uuid7_end, datetime_to_uuid7

# Partition-aware filtering
start = uuid7_start(datetime_to_uuid7(date_from))
end = uuid7_end(datetime_to_uuid7(date_to), settings.FINDINGS_TABLE_PARTITION_MONTHS)
queryset.filter(id__gte=start, id__lt=end)

from api.db_utils import batch_delete, create_objects_in_batches, update_objects_in_batches

# Delete in batches (RLS-aware)
batch_delete(tenant_id, queryset, batch_size=1000)

# Bulk create with RLS
create_objects_in_batches(tenant_id, Finding, objects, batch_size=500)

# Bulk update with RLS
update_objects_in_batches(tenant_id, Finding, objects, fields=["status"], batch_size=500)

Pattern	Rule
RLS in ViewSets	Automatic via `BaseRLSViewSet` - tenant_id from JWT
RLS in Celery	MUST use `@set_tenant` + `rls_transaction(tenant_id)`
Cross-tenant validation	Defense-in-depth: verify `obj.tenant_id == request.tenant_id`
Never trust user input	Use `request.tenant_id` from JWT, never `request.data.get("tenant_id")`
Admin DB bypass	Only for cross-tenant admin ops - exposes ALL tenants' data

Pattern	Rule
Named tasks only	NEVER use dynamic task names from user input
Validate arguments	Check UUID format before database queries
Safe queuing	Use `transaction.on_commit()` to enqueue AFTER commit
Modern retries	Use `autoretry_for`, `retry_backoff`, `retry_jitter`
Time limits	Set `soft_time_limit` and `time_limit` to prevent hung tasks
Idempotency	Use `update_or_create` or idempotency keys

# Safe task queuing - task only enqueued after transaction commits
with transaction.atomic():
    provider = Provider.objects.create(**data)
    transaction.on_commit(
        lambda: verify_provider_connection.delay(
            tenant_id=str(request.tenant_id),
            provider_id=str(provider.id)
        )
    )

# Modern retry pattern
@shared_task(
    base=RLSTask,
    bind=True,
    autoretry_for=(ConnectionError, TimeoutError, OperationalError),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
    max_retries=5,
    soft_time_limit=300,
    time_limit=360,
)
@set_tenant
def sync_provider_data(self, tenant_id, provider_id):
    with rls_transaction(tenant_id):
        # ... task logic
        pass

# Idempotent task - safe to retry
@shared_task(base=RLSTask, acks_late=True)
@set_tenant
def process_finding(tenant_id, finding_uid, data):
    with rls_transaction(tenant_id):
        Finding.objects.update_or_create(uid=finding_uid, defaults=data)

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

Setting	Production Value	Risk if Wrong
`DEBUG`	`False`	Exposes stack traces, settings, SQL queries
`SECRET_KEY`	Env var, rotated	Session hijacking, CSRF bypass
`ALLOWED_HOSTS`	Explicit list	Host header attacks
`SECURE_SSL_REDIRECT`	`True`	Credentials sent over HTTP
`SESSION_COOKIE_SECURE`	`True`	Session cookies over HTTP
`CSRF_COOKIE_SECURE`	`True`	CSRF tokens over HTTP
`SECURE_HSTS_SECONDS`	`31536000` (1 year)	Downgrade attacks
`CONN_MAX_AGE`	`60` or higher	Connection pool exhaustion

# Development
cd api && poetry run python src/backend/manage.py runserver
cd api && poetry run python src/backend/manage.py shell

# Celery
cd api && poetry run celery -A config.celery worker -l info -Q scans,overview
cd api && poetry run celery -A config.celery beat -l info

# Testing
cd api && poetry run pytest -x --tb=short

# Production checks
cd api && poetry run python src/backend/manage.py check --deploy

Library	Context7 ID	Use For
Celery	`/websites/celeryq_dev_en_stable`	Task patterns, queues, error handling
django-celery-beat	`/celery/django-celery-beat`	Periodic task scheduling
Django	`/websites/djangoproject_en_5_2`	Models, ORM, constraints, indexes

mcp_context7_query-docs(libraryId="/websites/celeryq_dev_en_stable", query="shared_task decorator retry patterns")
mcp_context7_query-docs(libraryId="/celery/django-celery-beat", query="periodic task database scheduler")
mcp_context7_query-docs(libraryId="/websites/djangoproject_en_5_2", query="model constraints CheckConstraint UniqueConstraint")

Database	Alias	Purpose	RLS
`default`	`prowler_user`	Standard API queries	Yes
`admin`	`admin`	Migrations, auth bypass	No
`replica`	`prowler_user`	Read-only queries	Yes
`admin_replica`	`admin`	Admin read replica	No

Primitive	Use For
`chain()`	Sequential execution: A → B → C
`group()`	Parallel execution: A, B, C simultaneously
Combined	Chain with nested groups for complex workflows

Mode	`tenant_id` in kwargs	`tenant_id` passed to function
`@set_tenant` (default)	Popped (removed)	NO - function doesn't receive it
`@set_tenant(keep_tenant=True)`	Read but kept	YES - function receives it

prowler-api

When to Use

Critical Rules

Related Skills

flow

verify

feature-flags

flags

When to Use

Critical Rules

Architecture Overview

4-Database Architecture

RLS Transaction Flow

Implementation Checklist

Decision Trees

Which Base Model?

Which Manager?

Which Database?

Celery Task Decorator Order?

RLS Model Pattern

M2M Relationships (MUST use through models)

Async Task Response Pattern (202 Accepted)

Providers (11 Supported)

RBAC Permissions

RBAC Visibility Pattern

Celery Queues

Task Composition (Canvas)

Beat Scheduling (Periodic Tasks)

Advanced Task Patterns

@set_tenant Behavior

Key Patterns

Celery Configuration

UUIDv7 for Partitioned Tables

Batch Operations with RLS

Security Patterns

Tenant Isolation Summary

Celery Task Security Summary

Quick Reference

Production Deployment Checklist

Critical Settings

Commands

Resources

Local References

Related Skills

Context7 MCP (Recommended)

Related Skills

flow

verify

feature-flags

flags

`@set_tenant` Behavior