prowler-api

适用场景

在以下 Prowler 专属模式中使用此技能：

• 行级安全（RLS）/ 租户隔离
• RBAC 权限和角色检查
• Provider 生命周期和验证
• 携带租户上下文的 Celery 任务
• 多数据库架构（4 数据库设置）

对于通用 DRF 模式（ViewSet、Serializer、Filter、JSON:API），请使用 django-drf 技能。

---

关键规则

• 在 ViewSet 上下文之外查询时，始终使用 rls_transaction(tenant_id)
• 检查权限前，始终使用 get_role()（只返回第一个角色）
• 始终按 @set_tenant 再 @handle_provider_deletion 的顺序使用装饰器
• 对 M2M 关系，始终使用显式 through 模型（RLS 要求如此）
• Celery 任务中绝不在没有 RLS 上下文的情况下访问 Provider.objects
• 绝不通过原始 SQL 或 connection.cursor() 绕过 RLS
• 绝不使用 Django 默认的 M2M；RLS 要求使用带有 tenant_id 的 through 模型

注意：rls_transaction() 接受 UUID 对象和字符串，并通过 str(value) 在内部进行转换。

---

架构概览

4 数据库架构

数据库	别名	用途	RLS
default	prowler_user	标准 API 查询	是
admin	admin	迁移、绕过身份验证	否
replica	prowler_user	只读查询	是
admin_replica	admin	管理员只读副本	否

# 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 事务流程

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

---

实现清单

实现 Prowler 专属 API 功能时：

#	模式	参考	要点
1	RLS 模型	api/rls.py	继承 RowLevelSecurityProtectedModel，添加约束
2	RLS 事务	api/db_utils.py	使用 rls_transaction(tenant_id) 上下文管理器
3	RBAC 权限	api/rbac/permissions.py	get_role()、get_providers()、Permissions 枚举
4	Provider 验证	api/models.py	在 Provider 模型上定义 validate_<provider>_uid() 方法
5	Celery 任务	tasks/tasks.py、api/decorators.py、config/celery.py	任务定义、装饰器（@set_tenant、@handle_provider_deletion）、RLSTask 基类
6	RLS Serializer	api/v1/serializers.py	继承 RLSSerializer，自动注入 tenant_id
7	Through 模型	api/models.py	所有 M2M 都必须使用带有 tenant_id 的显式 through 模型

完整文件路径：请参阅 references/file-locations.md

---

决策树

使用哪个基类模型？

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

使用哪个 Manager？

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

Celery 任务装饰器顺序？

@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 模型模式

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 关系（必须使用 through 模型）

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"],
            ),
        ]

---

异步任务响应模式（202 Accepted）

对于长时间运行的操作，返回 202 并附带任务引用：

@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（支持 11 种）

Provider	UID 格式	示例
AWS	12 位数字	123456789012
Azure	UUID v4	a1b2c3d4-e5f6-...
GCP	6–30 个字符，小写，以字母开头	my-gcp-project
M365	有效域名	contoso.onmicrosoft.com
Kubernetes	2–251 个字符	arn:aws:eks:...
GitHub	1–39 个字符	my-org
IaC	Git URL	https://github.com/user/repo.git
Oracle Cloud	OCID 格式	ocid1.tenancy.oc1..
MongoDB Atlas	24 位十六进制	507f1f77bcf86cd799439011
Alibaba Cloud	16 位数字	1234567890123456

添加新 Provider：添加到 ProviderChoices 枚举，并创建 validate_<provider>_uid() 静态方法。

---

RBAC 权限

权限	控制内容
MANAGE_USERS	用户 CRUD、角色分配
MANAGE_ACCOUNT	租户设置
MANAGE_BILLING	账单/订阅
MANAGE_PROVIDERS	Provider CRUD
MANAGE_INTEGRATIONS	集成配置
MANAGE_SCANS	扫描执行
UNLIMITED_VISIBILITY	查看所有 Provider（绕过 provider_groups）

RBAC 可见性模式

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 队列

队列	用途
scans	Prowler 扫描执行
overview	仪表板聚合（严重性、攻击面）
compliance	合规报告生成
integrations	外部集成（Jira、S3、Security Hub）
deletion	提供商/租户删除（异步）
backfill	历史数据回填操作
scan-reports	输出生成（CSV、JSON、HTML、PDF）

---

任务组合（Canvas）

使用 Celery 的 Canvas 原语构建复杂工作流：

原语	用途
chain()	顺序执行：A → B → C
group()	并行执行：同时执行 A、B、C
组合	使用嵌套 group 的 chain 构建复杂工作流

注意： 使用 .si()（签名不可变）可阻止传递结果。需要传递结果时使用 .s()。

示例： 查看 assets/celery_patterns.py，了解 chain、group 和组合模式。

---

Beat 调度（周期性任务）

操作	要点
创建计划	IntervalSchedule.objects.get_or_create(every=24, period=HOURS)
创建周期性任务	使用任务名称（而非函数），kwargs=json.dumps(...)
删除计划任务	PeriodicTask.objects.filter(name=...).delete()
避免竞态条件	使用 countdown=5 等待数据库提交

示例： 查看 assets/celery_patterns.py，了解 schedule_provider_scan 模式。

---

高级任务模式

@set_tenant 行为

模式	kwargs 中的 tenant_id	传递给函数的 tenant_id
@set_tenant（默认）	弹出（移除）	否——函数不会收到它
@set_tenant(keep_tenant=True)	读取但保留	是——函数会收到它

关键模式

模式	描述
bind=True	访问 self.request.id、self.request.retries
get_task_logger(__name__)	在 Celery 任务中正确记录日志
SoftTimeLimitExceeded	捕获该异常，以便在强制终止前保存进度
countdown=30	延迟 N 秒执行
eta=datetime(...)	在指定时间执行

示例： 查看 assets/celery_patterns.py，了解所有高级模式。

---

Celery 配置

设置	值	用途
BROKER_VISIBILITY_TIMEOUT	86400（24 小时）	防止长时间任务重新入队
CELERY_RESULT_BACKEND	django-db	将结果存储在 PostgreSQL 中
CELERY_TASK_TRACK_STARTED	True	跟踪任务的启动时间
soft_time_limit	任务专用	引发 SoftTimeLimitExceeded
time_limit	任务专用	强制终止（SIGKILL）

完整配置： 查看 assets/celery_patterns.py 以及 config/celery.py、config/settings/celery.py 中的实际文件。

---

分区表中的 UUIDv7

Finding 和 ResourceFindingMapping 使用 UUIDv7 进行基于时间的分区：

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)

为什么使用 UUIDv7？ 按时间排序的 UUID 可使 PostgreSQL 在范围查询期间裁剪分区。

---

使用 RLS 的批量操作

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)

---

安全模式

完整示例：查看 assets/security_patterns.py

租户隔离摘要

模式	规则
ViewSet 中的 RLS	通过 BaseRLSViewSet 自动实现——从 JWT 获取 tenant_id
Celery 中的 RLS	必须使用 @set_tenant + rls_transaction(tenant_id)
跨租户验证	深度防御：验证 obj.tenant_id == request.tenant_id
绝不信任用户输入	使用 JWT 中的 request.tenant_id，绝不使用 request.data.get("tenant_id")
管理员 DB 绕过	仅用于跨租户管理员操作——会暴露所有租户的数据

Celery 任务安全摘要

模式	规则
仅使用命名任务	切勿使用来自用户输入的动态任务名称
验证参数	在执行数据库查询前检查 UUID 格式
安全排队	使用 transaction.on_commit() 在提交后入队
现代重试机制	使用 autoretry_for、retry_backoff、retry_jitter
时间限制	设置 soft_time_limit 和 time_limit，防止任务挂起
幂等性	使用 update_or_create 或幂等键

快速参考

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

---

生产环境部署检查清单

完整设置：请参阅 references/production-settings.md

每次部署到生产环境前运行：

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

关键设置

设置	生产环境值	配置错误时的风险
DEBUG	False	暴露堆栈跟踪、设置和 SQL 查询
SECRET_KEY	环境变量，定期轮换	会话劫持、绕过 CSRF
ALLOWED_HOSTS	明确的列表	Host 标头攻击
SECURE_SSL_REDIRECT	True	凭据通过 HTTP 发送
SESSION_COOKIE_SECURE	True	会话 Cookie 通过 HTTP 发送
CSRF_COOKIE_SECURE	True	CSRF 令牌通过 HTTP 发送
SECURE_HSTS_SECONDS	31536000（1 年）	降级攻击
CONN_MAX_AGE	60 或更高	连接池耗尽

---

命令

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

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

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

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

---

资源

本地参考资料

• 文件位置：请参阅 references/file-locations.md
• 建模决策：请参阅 references/modeling-decisions.md
• 配置：请参阅 references/configuration.md
• 生产环境设置：请参阅 references/production-settings.md
• 安全模式：请参阅 assets/security_patterns.py

相关技能

• 通用 DRF 模式：使用 django-drf 技能
• API 测试：使用 prowler-test-api 技能

Context7 MCP（推荐）

**前提条件：**安装 Context7 MCP 服务，以便查询最新文档。

实现或调试 Prowler 专属模式时，通过 mcp_context7_query-docs 查询以下库：

库	Context7 ID	用途
Celery	/websites/celeryq_dev_en_stable	任务模式、队列、错误处理
django-celery-beat	/celery/django-celery-beat	定期任务调度
Django	/websites/djangoproject_en_5_2	模型、ORM、约束、索引

查询示例：

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

**注意：**如果需要查找正确的库 ID，请先使用 mcp_context7_resolve-library-id。