/prowler-api

name: prowler-api
description: >
  Prowler API patterns: RLS, RBAC, providers, Celery tasks, SSE.
  Trigger: When working in api/ on models/serializers/viewsets/filters/tasks involving tenant isolation (RLS), RBAC, provider lifecycle, Celery tasks or SSE endpoints.
license: Apache-2.0
metadata:
  author: prowler-cloud
  version: "1.3.0"
  scope: [root, api]
  auto_invoke: "Creating/modifying models, views, serializers"
allowed-tools: Read, Edit, Write, Glob, Grep, Bash, WebFetch, WebSearch, Task

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 (primary, replica, admin)
• Server-Sent Events (SSE) endpoints

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

All paths below are relative to `api/src/backend/` unless they start with `api/`.

---

Critical Rules

• ALWAYS use `rls_transaction(tenant_id)` when querying tenant data outside a ViewSet (tasks, jobs, management commands)
• ALWAYS take `tenant_id` from the auth token (`request.tenant_id`), NEVER from `request.data`
• ALWAYS pass `tenant_id` to `get_role(user, tenant_id)`; inside `BaseRLSViewSet` use the cached `self.user_role`
• ALWAYS use explicit through models (inheriting `RowLevelSecurityProtectedModel`) for M2M relationships
• ALWAYS give Celery tasks a `name=`; route with `queue=` (omitted = default `celery` queue)
• ALWAYS call tasks with keyword arguments (`tenant_id=...`): the decorators read `kwargs`
• NEVER access `Provider.objects` (or any tenant model) in Celery without RLS context
• NEVER combine plain `@set_tenant` with `@handle_provider_deletion`: `@set_tenant` pops `tenant_id`, which `@handle_provider_deletion` needs. Use `@set_tenant(keep_tenant=True)`
• NEVER use `MainRouter.admin_db` unless it is an auth or cross-tenant admin operation (it bypasses RLS)

> `rls_transaction(value, parameter=POSTGRES_TENANT_VAR, using=None, retry_on_replica=True)` validates that `value` is a UUID (UUID objects and strings both work) and raises `ValidationError` otherwise.

---

Architecture Overview

Databases (`config/django/{devel,production,testing}.py`)

| Alias | DB user | Purpose | RLS | |-------|---------|---------|-----| | `default` | `POSTGRES_USER` (`default` = the `prowler_user` entry) | Standard API queries | **Yes** | | `admin` | `POSTGRES_ADMIN_USER` | Migrations, auth lookups, Django internal tables | No | | `replica` | `POSTGRES_REPLICA_USER` | Read-only queries | **Yes** | | `admin_replica` | `POSTGRES_ADMIN_USER` | Admin reads on replica | No | | `neo4j` | `NEO4J_USER` | Graph DB for Attack Paths (not routed by Django) | - |

from api.db_router import MainRouter, READ_REPLICA_ALIAS

User.objects.using(MainRouter.admin_db).get(id=user_id)  # Auth lookups (bypasses RLS)

with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS):  # Read-heavy task code
    Finding.objects.filter(scan_id=scan_id).count()

• `MainRouter.allow_migrate()` only allows `admin`: run `manage.py migrate --database admin`.
• `READ_REPLICA_ALIAS` is `None` without a replica; `rls_transaction` then uses `default`.
• Replica failures retry (`POSTGRES_REPLICA_MAX_ATTEMPTS`, default 3) and then fall back to primary.

RLS Request Flow (`api/base_views.py`)

Request -> BaseRLSViewSet.dispatch()  replica for SAFE_METHODS (if configured), transaction.atomic
        -> BaseRLSViewSet.initial()   tenant_id from request.auth (JWT or API key)
                                      rls_transaction(tenant_id) -> set_config('api.tenant_id', ...)
        -> queries are tenant-scoped; serializer context gets tenant_id

Other bases in the same file: `BaseTenantViewset` (tenant CRUD), `BaseUserViewset` (user endpoints), `BaseViewSet` (no RLS).

---

Implementation Checklist

| # | Pattern | Reference | Key Points | |---|---------|-----------|------------| | 1 | **RLS Models** | `api/rls.py` | Inherit `RowLevelSecurityProtectedModel`, add `RowLevelSecurityConstraint` | | 2 | **RLS Transactions** | `api/db_utils.py` | `rls_transaction(tenant_id)` context manager | | 3 | **RBAC Permissions** | `api/rbac/permissions.py` | `Permissions` enum, `HasPermissions`, `get_role()`, `get_providers()` | | 4 | **Provider Validation** | `api/models.py` | `validate_<provider>_uid()` staticmethods on `Provider` | | 5 | **Celery Tasks** | `tasks/tasks.py`, `api/decorators.py`, `config/celery.py` | `@shared_task(base=RLSTask, name=...)`, `@set_tenant`, `@handle_provider_deletion` | | 6 | **Task business logic** | `tasks/jobs/*.py` | Keep `tasks.py` thin; logic lives in `jobs/` | | 7 | **RLS Serializers** | `api/v1/serializers.py` | Inherit `RLSSerializer` to auto-inject `tenant_id` | | 8 | **Base ViewSets** | `api/base_views.py` | `BaseRLSViewSet` for tenant-scoped resources | | 9 | **SSE endpoints** | `api/sse/` | Subclass `BaseSSEViewSet`, build channels with `make_channel_name()` | | 10 | **Changelog** | `api/changelog.d/` | Add a towncrier fragment, never edit `CHANGELOG.md` (see `prowler-changelog`) |

> **Full file paths**: See [references/file-locations.md](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
Hide data of deleted providers -> objects = ActiveProviderManager(); all_objects = models.Manager()

Which Manager?

Normal queries           -> Model.objects (ActiveProviderManager filters out deleted providers)
Include deleted records  -> Model.all_objects
Celery task context      -> Must be inside rls_transaction() first

Which Database?

Standard API queries     -> default (automatic via V