Prowler API 多租户安全架构实战：RLS 租户隔离、RBAC 权限与带租户上下文的 Celery 任务

Prowler API 多租户安全架构实战：RLS 租户隔离、RBAC 权限与带租户上下文的 Celery 任务

【免费下载链接】prowlerProwler is the world’s most widely used open-source cloud security platform that automates security and compliance across any cloud environment.项目地址: https://gitcode.com/GitHub_Trending/pr/prowler

Prowler 的 API 后端（api/目录）是一套面向多租户 SaaS 的 Django + Celery 架构：所有云账号（Provider）扫描、发现项（Finding）存储与合规计算都建立在严格的租户隔离之上。本文基于仓库内的技能文档skills/prowler-api/SKILL.md及其引用的源码，系统讲解四库（4-database）架构、PostgreSQL 行级安全（RLS）、RBAC 权限模型、Provider 生命周期校验、带租户上下文的 Celery 任务模式，以及生产部署检查清单。读完本文，你可以在不破坏租户隔离的前提下为 Prowler API 新增模型、任务与权限控制，并理解其底层 SQL 策略与故障回退机制。

适用场景与总体规则

该技能文档明确了 Prowler 特有模式与通用 DRF 模式的分工：涉及租户隔离（RLS）、RBAC 权限与角色检查、Provider 生命周期校验、带租户上下文的 Celery 任务、多数据库架构时适用本技能；而 ViewSets、Serializers、Filters、JSON:API 等通用模式则交给skills/django-drf/SKILL.md。

文档给出了一组必须遵守的关键规则（Critical Rules）：

• 在 ViewSet 上下文之外查询数据时，必须使用rls_transaction(tenant_id)；
• 权限检查前必须先调用get_role()（它只返回用户在该租户下的第一个角色）；
• Celery 任务上@set_tenant装饰器必须位于@handle_provider_deletion之前（更靠近函数体）；
• 所有多对多关系必须使用显式 through 模型（RLS 要求 through 表带tenant_id），禁止使用 Django 默认 M2M；
• 在 Celery 任务中，未经 RLS 上下文直接访问Provider.objects是禁止的；
• 严禁通过裸 SQL 或connection.cursor()绕过 RLS。

需要说明的一点：rls_transaction()同时接受 UUID 对象与字符串，内部通过str(value)转换；从 db_utils.py 的源码看，它在设置租户变量前会先用uuid.UUID(str(value))校验，非法 UUID 直接抛出ValidationError("Must be a valid UUID")。

四数据库架构（4-Database Architecture）

# 何时使用 admin（绕过 RLS） from api.db_router import MainRouter User.objects.using(MainRouter.admin_db).get(id=user_id) # 鉴权查询 # 标准查询走 default（强制 RLS） Provider.objects.filter(connected=True) # 需要 rls_transaction 上下文

从 db_router.py 的MainRouter源码可以确认其路由逻辑：db_for_read中，凡是表名以django_或socialaccount_、account_、authtoken_、silk_开头的模型（即 Django 认证/社交登录等基础设施表）一律路由到admin库——这正是"鉴权旁路"的实现；业务模型则跟随ContextVar中记录的读别名（get_read_db_alias()），默认为None（回落到 Django 默认的default）。allow_migrate只在admin库返回True，保证迁移由管理员账号执行，而prowler_user这个受 RLS 约束的账号只持有最少权限。

此外，configuration.md 中的DATABASES定义还包含prowler_user（RLS 连接的原始定义）和neo4j（攻击路径用的图数据库）两个别名；replica、admin_replica通过POSTGRES_REPLICA_HOST等环境变量选配，未配置时副本功能自动关闭。

RLS 事务流：一个请求的租户上下文如何注入

文档给出的 RLS 事务流程图：

Request → Authentication → BaseRLSViewSet.initial() │ ├─ 从 JWT 中提取 tenant_id ├─ SET api.tenant_id = 'uuid' (PostgreSQL) └─ 之后所有查询都被限定到该租户

从源码看，这条链路的落点非常清晰。rls_transaction在 db_utils.py 中定义，其核心动作是执行一条 PostgreSQL 事务级配置语句：

SET_CONFIG_QUERY = "SELECT set_config(%s, %s::text, TRUE);" POSTGRES_TENANT_VAR = "api.tenant_id"

set_config的第三个参数TRUE表示该设置仅在当前事务内生效，事务结束即失效——这意味着租户上下文绝不会跨请求/跨任务泄漏，连接池复用也是安全的。

而真正执行隔离的是 rls.py 中RowLevelSecurityConstraint在迁移时生成的策略 SQL。每个受保护表会执行：

ALTER TABLE <table> ENABLE ROW LEVEL SECURITY; ALTER TABLE <table> FORCE ROW LEVEL SECURITY; CREATE POLICY <db_user>_<table>_<statement> ON <table> FOR <statement> TO <db_user> USING ( CASE WHEN current_setting('api.tenant_id', True) IS NULL THEN FALSE ELSE <tenant_column> = current_setting('api.tenant_id')::uuid END );

这里有三个值得注意的设计：

1. FORCE ROW LEVEL SECURITY：连表属主也受策略约束；
2. IS NULL THEN FALSE：租户变量未设置时策略直接拒绝所有行——即使有人忘记了 RLS 上下文，结果也是"看不到任何数据"而非"看到全部数据"，这是 fail-closed 设计；
3. 策略只授予prowler_user（API 数据库账号）指定语句的权限，INSERT用WITH CHECK，其余语句用USING，形成"最小权限 + 行级策略"双保险。对全局/共享数据，则使用同文件中的BaseSecurityConstraint（rls.py），只授予最小权限而不启用 RLS。

rls_transaction还有一个文档未展开但源码中很完整的能力——副本故障回退（见 db_utils.py 的 docstring）：当using指向只读副本时，连接建立失败会按POSTGRES_REPLICA_MAX_ATTEMPTS（默认 3 次）重试并以指数退避（基数POSTGRES_REPLICA_RETRY_BASE_DELAY，默认 0.5 秒）延迟，最终回落到主库；查询中途的连接级OperationalError则由execute_wrapper拦截，仅对"单条纯 SELECT 且无锁子句"的安全语句在主库上以只读事务重放，死锁/序列化失败/用户取消这类错误则原样抛给调用方，避免掩盖真实并发问题。源码同时注明了限制：服务端游标（.iterator()）的行拉取不会被拦截，大规模迭代需自行重试。

RLS 模型模式：租户表怎么写

文档给出的标准模型模板：

from api.rls import RowLevelSecurityProtectedModel, RowLevelSecurityConstraint class MyModel(RowLevelSecurityProtectedModel): # tenant FK 从父类继承 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"

与源码对照：rls.py 中RowLevelSecurityProtectedModel是抽象基类，继承自models.Model并自带tenant = models.ForeignKey("Tenant", on_delete=models.CASCADE)；Tenant模型本身（UUID 主键 +name，表名tenants）是整个系统的"基本分组"，用于在不同客户之间隔离数据。约束类的validate()方法（rls.py）会在模型校验时检查实例必须含tenant_id字段。另外约束支持partition_name参数（见 rls.py 的create_sql），可以把 RLS 策略直接应用到findings_2025_aug这类分区表上——这与后文的 UUIDv7 按月分区是配套设计。

多对多关系必须显式声明 through 模型

class Resource(RowLevelSecurityProtectedModel): tags = models.ManyToManyField( ResourceTag, through="ResourceTagMapping", # RLS 必需 ) class ResourceTagMapping(RowLevelSecurityProtectedModel): # through 模型必须带 tenant_id 才能启用 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"], ), ]

原因很直接：Django 自动生成的 M2M 中间表没有tenant_id列，无法为其创建基于租户的策略，也就无法阻止跨租户关联。文档同时给出了选型决策树：

• 选哪个基类模型？租户级数据 →RowLevelSecurityProtectedModel；全局/共享数据 →models.Model+BaseSecurityConstraint（少见）；分区时序数据 →PostgresPartitionedModel+RowLevelSecurityProtectedModel；软删除 → 追加is_deleted字段 +ActiveProviderManager。
• 选哪个 Manager？常规查询用Model.objects（排除已删除）；需要已删除记录用Model.all_objects（models.py 中Provider即定义了all_objects = models.Manager()）；Celery 任务上下文必须先rls_transaction()。
• 选哪个库？标准 API 查询走default（ViewSet 自动）；只读操作走replica（BaseRLSViewSet对 GET 自动）；鉴权/管理操作走MainRouter.admin_db；跨租户查询走admin库（谨慎使用）。
• Celery 装饰器顺序？@shared_task(base=RLSTask, ...)之下先@set_tenant（设置租户上下文），再@handle_provider_deletion（处理扫描期间被删除的 Provider）。

异步任务响应模式（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})} )

这里的Task是业务侧的任务模型。从 celery.py 的RLSTask源码可以看到闭环：RLSTask.apply_async在任务派发后，用kwargs里的tenant_id打开rls_transaction，在api.models.Task表中update_or_create出一条业务任务记录并关联 Celery 的TaskResult——所以任务一入队，租户内就能通过 API 查到任务对象及其状态，结果后端django-db把结果存在 PostgreSQL 而非独立缓存。

Provider 生命周期与 UID 校验

文档列出的 Provider 与 UID 格式表（"Adding new provider"：向ProviderChoices枚举追加成员，并实现对应的validate_<provider>_uid()静态方法）：

对照 models.py 的ProviderChoices源码，当前枚举实际上已扩展到 16 个成员，在文档表格基础上还包含 Cloudflare、OpenStack、Image、Google Workspace、Vercel、Okta（文档标题写 "11 Supported" 而表格仅 10 行，可视为技能文档滞后于代码的例证）。源码中每个校验方法的实现细节也与表格吻合，例如：

• validate_aws_uid：re.match(r"^\d{12}$", value)，失败抛出带 JSON:APIpointer="/data/attributes/uid"的ModelValidationError（models.py）；
• validate_azure_uid：要求是严格 UUID v4 且字符串形式与规范化形式一致（models.py）；
• validate_gcp_uid：6-30 字符、字母开头，另兼容domain.com:project-id形式的旧版 App Engine 项目 ID（models.py）；
• validate_kubernetes_uid：接受 K8s UID、AWS EKS ARN、GKE Context 名或 Azure AKS 集群名（models.py）。

这些错误统一以 JSON:API 错误指针（/data/attributes/uid）返回，与 configuration.md 中JSON_API_UNIFORM_EXCEPTIONS: True的全局异常格式相呼应。

RBAC 权限模型

文档的权限表：

从 permissions.py 看，这正是Permissions枚举的 7 个成员。文档给出的可见性过滤模式：

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: # 按角色分配的 provider_groups 过滤 return Model.objects.filter(provider__in=get_providers(user_role))

源码印证了两处关键细节：

• get_role(user, tenant_id)（permissions.py）通过User.roles.using(MainRouter.admin_db).filter(tenant_id=tenant_id).first()取第一个角色（用户-角色关联表本身在 admin 库中），无角色时抛PermissionDenied——这就是技能文档反复强调"先get_role()再判断"且"只返回第一个角色"的原因；
• get_providers(role)（permissions.py）按角色关联的 provider 分组返回去重后的 Provider 查询集，角色没有任何分组时返回空集（即什么也看不到）。

视图层则统一用HasPermissions基类（permissions.py）：从视图属性required_permissions读取所需权限列表，再对该用户在此租户下的所有角色做"任一角色具备该权限"的聚合判定——这与get_role()的单角色语义形成互补：HasPermissions负责"能不能做这个操作"，get_role+get_providers负责"能看到哪些数据"。

Celery 任务体系：队列、装饰器与 Canvas

队列划分

按 file-locations.md 的路径表，任务定义集中在api/src/backend/tasks/tasks.py，业务逻辑分层在tasks/jobs/下：scan.py（perform_prowler_scan()、aggregate_findings()）、deletion.py（delete_provider()、delete_tenant()）、export.py（CSV/JSON/HTML）、report.py（PDF 报告）、integrations.py（S3/Security Hub/Jira 上传）、attack_paths/（Neo4j/Cartography 攻击路径）等。

两个核心装饰器

@set_tenant的两种模式（文档表格）：

从 decorators.py 源码看，set_tenant的行为比表格更完整：它先用@transaction.atomic包住整个任务，校验tenant_id是合法 UUID（否则ValidationError），再通过set_config('api.tenant_id', ...)在当前连接上设置租户变量——与 ViewSet 请求路径用的是同一条机制，从而保证"请求上下文"与"任务上下文"共享同一套 RLS 语义。

@handle_provider_deletion（decorators.py）处理"扫描执行到一半 Provider 被删"的竞态：捕获ObjectDoesNotExist/DatabaseError/GraphDatabaseQueryException后，在rls_transaction内回查 Provider 是否还存在，不存在则转抛ProviderDeletedException；若任务 kwargs 里只有scan_id，会先经 Scan 反查provider_id；对图数据库异常还会额外校验租户与 Membership 是否仍存在。

任务编写范式

@shared_task(base=RLSTask, name="task-name", queue="scans") @set_tenant # 先：设置租户上下文 @handle_provider_deletion # 后：处理被删除的 Provider def my_task(tenant_id: str, provider_id: str): with rls_transaction(tenant_id): provider = Provider.objects.get(pk=provider_id)

文档推荐的关键任务模式：

配套的"安全任务"参考实现（文档 Quick Reference）：

# 安全的任务入队 —— 事务提交后才入队 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) ) ) # 现代重试模式 @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): # ... 任务逻辑 pass # 幂等任务 —— 重试安全 @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)

复杂工作流：Canvas 原语

注意.si()（签名不可变）用于阻止结果传递，需要用.s()时才传递结果；链式/group 示例见 assets/celery_patterns.py。

定时任务（django-celery-beat）

schedule_provider_scan()的完整示例在 tasks/beat.py 与 assets/celery_patterns.py。

Celery 关键配置

celery.py 的源码把这些配置落实得很具体：broker 是 Valkey/Redis（CELERY_BROKER_URL由VALKEY_*环境变量拼装），visibility_timeout默认 86400 秒（DJANGO_BROKER_VISIBILITY_TIMEOUT）；task_acks_late = True+task_reject_on_worker_lost = True+worker_prefetch_multiplier = 1组合成"持久投递"——worker 在任务中途被杀（部署/OOM/驱逐）时消息不会静默丢失，而会重新入队；worker_soft_shutdown_timeout默认 60 秒，让 SIGTERM 时有时间完成或重排未完成任务。时间上限则按任务分级：连接检查类任务（如provider-connection-check）用 60s/120s 的紧上限，扫描与删除类任务（scan-perform、provider-deletion、tenant-deletion等）用 48 小时上限（大租户的扫描和删除可能合法地运行一天以上），其余任务默认硬上限 6 小时（celery.py）。

UUIDv7 与按月分区

Finding和ResourceFindingMapping使用 UUIDv7 以支持按时间分区：

from uuid6 import uuid7 from api.uuid_utils import uuid7_start, uuid7_end, datetime_to_uuid7 # 分区感知的过滤 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 在处理范围查询时可以裁剪（prune）分区——主键即时间戳，id__gte/id__lt的范围条件能直接映射到分区边界。分区参数来自 configuration.md：FINDINGS_TABLE_PARTITION_MONTHS（默认 1，按月）、FINDINGS_TABLE_PARTITION_COUNT（默认 7）、可选的FINDINGS_TABLE_PARTITION_MAX_AGE_MONTHS（过期清理）；分区管理器实现在api/src/backend/api/partitions.py（PartitionManager），db_utils.py 中的_should_create_index_on_partition还解释了分区命名规则（findings_2025_aug形式）与"新索引默认只建在当前及未来分区"的策略，以降低旧分区维护开销。

带 RLS 的批量操作

from api.db_utils import batch_delete, create_objects_in_batches, update_objects_in_batches # 分批删除（RLS 感知） batch_delete(tenant_id, queryset, batch_size=1000) # 带 RLS 的批量创建 create_objects_in_batches(tenant_id, Finding, objects, batch_size=500) # 带 RLS 的批量更新 update_objects_in_batches(tenant_id, Finding, objects, fields=["status"], batch_size=500)

从 db_utils.py 的源码看，三个函数的共同点是"每一批各自包在一个rls_transaction里"，确保任何单个事务都不会膨胀过大；batch_delete的默认批大小取自DJANGO_DELETION_BATCH_SIZE（默认 5000），而create_objects_in_batches/update_objects_in_batches默认 500，分别走bulk_create/bulk_update。DJANGO_FINDINGS_BATCH_SIZE（默认 1000）则用于 Finding 导出场景。

安全模式汇总

文档总结的两张安全速查表：

租户隔离：

Celery 任务安全：

完整示例见 assets/security_patterns.py。

生产部署检查清单

每次生产部署前运行：

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

关键设置与风险（详见 references/production-settings.md）：

常用命令

# 开发 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 # 测试 cd api && uv run pytest -x --tb=short # 生产检查 cd api && uv run python src/backend/manage.py check --deploy

文件导航与延伸阅读

按 references/file-locations.md 的路径表，本文涉及的核心文件在仓库中的位置：

技能文档建议：通用 DRF 模式（ViewSets、Serializers、Filters、JSON:API）参考skills/django-drf/SKILL.md，API 测试模式参考skills/prowler-test-api/SKILL.md。至此，本文完整覆盖了 Prowler API 的多租户隔离（RLS 策略 + 事务级set_config）、RBAC 可见性、Provider 校验与 Celery 租户上下文这条主线——它们共同回答了一个核心问题：如何让一个云安全平台在共享 PostgreSQL 集群上安全地服务成千上万个互不可见的客户。

【免费下载链接】prowlerProwler is the world’s most widely used open-source cloud security platform that automates security and compliance across any cloud environment.项目地址: https://gitcode.com/GitHub_Trending/pr/prowler