prowler-api

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 connection.cursor()
• NEVER use Django's default M2M - RLS requires through models with tenant_id

Note: rls_transaction() accepts both UUID objects and strings - it converts internally via str(value).

---

Architecture Overview

4-Database Architecture

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

# 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

RLS Transaction Flow

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

---

Implementation Checklist

When implementing Prowler-specific API features:

#	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

Full file paths: See references/file-locations.md

---

Decision Trees

Which Base Model?

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

Which Manager?

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

Which Database?

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!)

Celery Task Decorator Order?

@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

---

RLS Model Pattern

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"

M2M Relationships (MUST use through 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"],
            ),
        ]

---

Async Task Response Pattern (202 Accepted)

For long-running operations, return 202 with task reference:

@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})}
    )

---

Providers (11 Supported)

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

Adding new provider: Add to ProviderChoices enum + create validate_<provider>_uid() staticmethod.

---

RBAC Permissions

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)

RBAC Visibility Pattern

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))

---

Celery Queues

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)

---

Task Composition (Canvas)

Use Celery's Canvas primitives for complex workflows:

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

Note: Use .si() (signature immutable) to prevent result passing. Use .s() if you need to pass results.

Examples: See assets/celery_patterns.py for chain, group, and combined patterns.

---

Beat Scheduling (Periodic Tasks)

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

Examples: See assets/celery_patterns.py for schedule_provider_scan pattern.

---

Advanced Task Patterns

@set_tenant Behavior

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

Key Patterns

Pattern	Description
bind=True	Access self.request.id, self.request.retries
get_task_logger(__name__)	Proper logging in Celery tasks
SoftTimeLimitExceeded

---

Content truncated.