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/prowlerProwler 的 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数据库别名用途是否启用 RLSdefaultprowler_user标准 API 查询是adminadmin迁移、鉴权旁路否replicaprowler_user只读查询是admin_replicaadmin管理端只读副本否# 何时使用 admin绕过 RLS from api.db_router import MainRouter User.objects.using(MainRouter.admin_db).get(iduser_id) # 鉴权查询 # 标准查询走 default强制 RLS Provider.objects.filter(connectedTrue) # 需要 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_userRLS 连接的原始定义和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_idset_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 );这里有三个值得注意的设计FORCE ROW LEVEL SECURITY连表属主也受策略约束IS NULL THEN FALSE租户变量未设置时策略直接拒绝所有行——即使有人忘记了 RLS 上下文结果也是看不到任何数据而非看到全部数据这是 fail-closed 设计策略只授予prowler_userAPI 数据库账号指定语句的权限INSERT用WITH CHECK其余语句用USING形成最小权限 行级策略双保险。对全局/共享数据则使用同文件中的BaseSecurityConstraintrls.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_keyTrue, defaultuuid4, editableFalse) name models.CharField(max_length255) inserted_at models.DateTimeField(auto_now_addTrue, editableFalse) updated_at models.DateTimeField(auto_nowTrue, editableFalse) class Meta(RowLevelSecurityProtectedModel.Meta): db_table my_models constraints [ RowLevelSecurityConstraint( fieldtenant_id, namerls_on_%(class)s, statements[SELECT, INSERT, UPDATE, DELETE], ), ] class JSONAPIMeta: resource_name my-models与源码对照rls.py 中RowLevelSecurityProtectedModel是抽象基类继承自models.Model并自带tenant models.ForeignKey(Tenant, on_deletemodels.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, throughResourceTagMapping, # RLS 必需 ) class ResourceTagMapping(RowLevelSecurityProtectedModel): # through 模型必须带 tenant_id 才能启用 RLS resource models.ForeignKey(Resource, on_deletemodels.CASCADE) tag models.ForeignKey(ResourceTag, on_deletemodels.CASCADE) class Meta: constraints [ RowLevelSecurityConstraint( fieldtenant_id, namerls_on_%(class)s, statements[SELECT, INSERT, UPDATE, DELETE], ), ]原因很直接Django 自动生成的 M2M 中间表没有tenant_id列无法为其创建基于租户的策略也就无法阻止跨租户关联。文档同时给出了选型决策树选哪个基类模型租户级数据 →RowLevelSecurityProtectedModel全局/共享数据 →models.ModelBaseSecurityConstraint少见分区时序数据 →PostgresPartitionedModelRowLevelSecurityProtectedModel软删除 → 追加is_deleted字段 ActiveProviderManager。选哪个 Manager常规查询用Model.objects排除已删除需要已删除记录用Model.all_objectsmodels.py 中Provider即定义了all_objects models.Manager()Celery 任务上下文必须先rls_transaction()。选哪个库标准 API 查询走defaultViewSet 自动只读操作走replicaBaseRLSViewSet对 GET 自动鉴权/管理操作走MainRouter.admin_db跨租户查询走admin库谨慎使用。Celery 装饰器顺序shared_task(baseRLSTask, ...)之下先set_tenant设置租户上下文再handle_provider_deletion处理扫描期间被删除的 Provider。异步任务响应模式202 Accepted长耗时操作的标准返回方式是 202 任务引用action(detailTrue, methods[post], url_nameconnection) def connection(self, request, pkNone): with transaction.atomic(): task check_provider_connection_task.delay( provider_idpk, tenant_idself.request.tenant_id ) prowler_task Task.objects.get(idtask.id) serializer TaskSerializer(prowler_task) return Response( dataserializer.data, statusstatus.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()静态方法ProviderUID 格式示例AWS12 位数字123456789012AzureUUID v4a1b2c3d4-e5f6-...GCP6-30 字符小写字母开头my-gcp-projectM365合法域名contoso.onmicrosoft.comKubernetes2-251 字符arn:aws:eks:...GitHub1-39 字符my-orgIaCGit URLhttps://github.com/user/repo.gitOracle CloudOCID 格式ocid1.tenancy.oc1..MongoDB Atlas24 位十六进制507f1f77bcf86cd799439011Alibaba Cloud16 位数字1234567890123456对照 models.py 的ProviderChoices源码当前枚举实际上已扩展到 16 个成员在文档表格基础上还包含 Cloudflare、OpenStack、Image、Google Workspace、Vercel、Okta文档标题写 11 Supported 而表格仅 10 行可视为技能文档滞后于代码的例证。源码中每个校验方法的实现细节也与表格吻合例如validate_aws_uidre.match(r^\d{12}$, value)失败抛出带 JSON:APIpointer/data/attributes/uid的ModelValidationErrormodels.pyvalidate_azure_uid要求是严格 UUID v4 且字符串形式与规范化形式一致models.pyvalidate_gcp_uid6-30 字符、字母开头另兼容domain.com:project-id形式的旧版 App Engine 项目 IDmodels.pyvalidate_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 权限模型文档的权限表权限控制范围MANAGE_USERS用户 CRUD、角色分配MANAGE_ACCOUNT租户设置MANAGE_BILLING计费/订阅MANAGE_PROVIDERSProvider CRUDMANAGE_INTEGRATIONS集成配置MANAGE_SCANS扫描执行UNLIMITED_VISIBILITY可见所有 Provider绕过 provider_groups从 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_idself.request.tenant_id) else: # 按角色分配的 provider_groups 过滤 return Model.objects.filter(provider__inget_providers(user_role))源码印证了两处关键细节get_role(user, tenant_id)permissions.py通过User.roles.using(MainRouter.admin_db).filter(tenant_idtenant_id).first()取第一个角色用户-角色关联表本身在 admin 库中无角色时抛PermissionDenied——这就是技能文档反复强调先get_role()再判断且只返回第一个角色的原因get_providers(role)permissions.py按角色关联的 provider 分组返回去重后的 Provider 查询集角色没有任何分组时返回空集即什么也看不到。视图层则统一用HasPermissions基类permissions.py从视图属性required_permissions读取所需权限列表再对该用户在此租户下的所有角色做任一角色具备该权限的聚合判定——这与get_role()的单角色语义形成互补HasPermissions负责能不能做这个操作get_roleget_providers负责能看到哪些数据。Celery 任务体系队列、装饰器与 Canvas队列划分队列用途scansProwler 扫描执行overview仪表盘聚合严重度、攻击面compliance合规报告生成integrations外部集成Jira、S3、Security HubdeletionProvider/租户删除异步backfill历史数据回填scan-reports输出文件生成CSV、JSON、HTML、PDF按 file-locations.md 的路径表任务定义集中在api/src/backend/tasks/tasks.py业务逻辑分层在tasks/jobs/下scan.pyperform_prowler_scan()、aggregate_findings()、deletion.pydelete_provider()、delete_tenant()、export.pyCSV/JSON/HTML、report.pyPDF 报告、integrations.pyS3/Security Hub/Jira 上传、attack_paths/Neo4j/Cartography 攻击路径等。两个核心装饰器set_tenant的两种模式文档表格模式kwargs 中的tenant_id函数是否收到tenant_idset_tenant默认弹出移除否set_tenant(keep_tenantTrue)读取但保留是从 decorators.py 源码看set_tenant的行为比表格更完整它先用transaction.atomic包住整个任务校验tenant_id是合法 UUID否则ValidationError再通过set_config(api.tenant_id, ...)在当前连接上设置租户变量——与 ViewSet 请求路径用的是同一条机制从而保证请求上下文与任务上下文共享同一套 RLS 语义。handle_provider_deletiondecorators.py处理扫描执行到一半 Provider 被删的竞态捕获ObjectDoesNotExist/DatabaseError/GraphDatabaseQueryException后在rls_transaction内回查 Provider 是否还存在不存在则转抛ProviderDeletedException若任务 kwargs 里只有scan_id会先经 Scan 反查provider_id对图数据库异常还会额外校验租户与 Membership 是否仍存在。任务编写范式shared_task(baseRLSTask, nametask-name, queuescans) set_tenant # 先设置租户上下文 handle_provider_deletion # 后处理被删除的 Provider def my_task(tenant_id: str, provider_id: str): with rls_transaction(tenant_id): provider Provider.objects.get(pkprovider_id)文档推荐的关键任务模式模式说明bindTrue访问self.request.id、self.request.retriesget_task_logger(__name__)Celery 任务中的正确日志方式SoftTimeLimitExceeded捕获后在硬杀前保存进度countdown30延迟 N 秒执行etadatetime(...)指定时间执行配套的安全任务参考实现文档 Quick Reference# 安全的任务入队 —— 事务提交后才入队 with transaction.atomic(): provider Provider.objects.create(**data) transaction.on_commit( lambda: verify_provider_connection.delay( tenant_idstr(request.tenant_id), provider_idstr(provider.id) ) ) # 现代重试模式 shared_task( baseRLSTask, bindTrue, autoretry_for(ConnectionError, TimeoutError, OperationalError), retry_backoffTrue, retry_backoff_max600, retry_jitterTrue, max_retries5, soft_time_limit300, time_limit360, ) set_tenant def sync_provider_data(self, tenant_id, provider_id): with rls_transaction(tenant_id): # ... 任务逻辑 pass # 幂等任务 —— 重试安全 shared_task(baseRLSTask, acks_lateTrue) set_tenant def process_finding(tenant_id, finding_uid, data): with rls_transaction(tenant_id): Finding.objects.update_or_create(uidfinding_uid, defaultsdata)复杂工作流Canvas 原语原语用途chain()顺序执行A → B → Cgroup()并行执行A、B、C 同时进行组合chain 内嵌 group构建复杂工作流注意.si()签名不可变用于阻止结果传递需要用.s()时才传递结果链式/group 示例见 assets/celery_patterns.py。定时任务django-celery-beat操作要点创建调度IntervalSchedule.objects.get_or_create(every24, periodHOURS)创建周期任务使用任务名而非函数kwargsjson.dumps(...)删除周期任务PeriodicTask.objects.filter(name...).delete()避免竞态用countdown5等待数据库提交schedule_provider_scan()的完整示例在 tasks/beat.py 与 assets/celery_patterns.py。Celery 关键配置设置值目的BROKER_VISIBILITY_TIMEOUT8640024h防止长任务被重新入队CELERY_RESULT_BACKENDdjango-db结果存 PostgreSQLCELERY_TASK_TRACK_STARTEDTrue跟踪任务开始soft_time_limit按任务设置抛出SoftTimeLimitExceededtime_limit按任务设置硬杀SIGKILLcelery.py 的源码把这些配置落实得很具体broker 是 Valkey/RedisCELERY_BROKER_URL由VALKEY_*环境变量拼装visibility_timeout默认 86400 秒DJANGO_BROKER_VISIBILITY_TIMEOUTtask_acks_late Truetask_reject_on_worker_lost Trueworker_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__gtestart, id__ltend)为什么用 UUIDv7时间有序的 UUID 让 PostgreSQL 在处理范围查询时可以裁剪prune分区——主键即时间戳id__gte/id__lt的范围条件能直接映射到分区边界。分区参数来自 configuration.mdFINDINGS_TABLE_PARTITION_MONTHS默认 1按月、FINDINGS_TABLE_PARTITION_COUNT默认 7、可选的FINDINGS_TABLE_PARTITION_MAX_AGE_MONTHS过期清理分区管理器实现在api/src/backend/api/partitions.pyPartitionManagerdb_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_size1000) # 带 RLS 的批量创建 create_objects_in_batches(tenant_id, Finding, objects, batch_size500) # 带 RLS 的批量更新 update_objects_in_batches(tenant_id, Finding, objects, fields[status], batch_size500)从 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 导出场景。安全模式汇总文档总结的两张安全速查表租户隔离模式规则ViewSet 中的 RLS经BaseRLSViewSet自动完成——tenant_id 来自 JWTCelery 中的 RLS必须set_tenantrls_transaction(tenant_id)跨租户校验纵深防御验证obj.tenant_id request.tenant_id永不信任用户输入用 JWT 中的request.tenant_id绝不用request.data.get(tenant_id)Admin 库旁路仅用于跨租户管理操作——它会暴露所有租户的数据Celery 任务安全模式规则只用命名任务绝不用来自用户输入的动态任务名校验参数数据库查询前检查 UUID 格式安全入队用transaction.on_commit()在提交后入队现代重试autoretry_for、retry_backoff、retry_jitter时间上限设置soft_time_limit与time_limit防止任务挂死幂等性update_or_create或幂等键完整示例见 assets/security_patterns.py。生产部署检查清单每次生产部署前运行cd api uv run python src/backend/manage.py check --deploy关键设置与风险详见 references/production-settings.md设置生产取值配置错误的风险DEBUGFalse暴露堆栈、设置与 SQLSECRET_KEY环境变量定期轮换会话劫持、CSRF 绕过ALLOWED_HOSTS显式列表Host 头攻击SECURE_SSL_REDIRECTTrue凭据走 HTTPSESSION_COOKIE_SECURETrue会话 Cookie 走 HTTPCSRF_COOKIE_SECURETrueCSRF Token 走 HTTPSECURE_HSTS_SECONDS315360001 年降级攻击CONN_MAX_AGE60或更高连接池耗尽常用命令# 开发 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 --tbshort # 生产检查 cd api uv run python src/backend/manage.py check --deploy文件导航与延伸阅读按 references/file-locations.md 的路径表本文涉及的核心文件在仓库中的位置主题文件RLS 基类模型与约束api/src/backend/api/rls.pyrls_transaction()、批量操作api/src/backend/api/db_utils.py四库路由MainRouterapi/src/backend/api/db_router.pyRBAC 权限api/src/backend/api/rbac/permissions.pyProvider 模型与 UID 校验api/src/backend/api/models.pyset_tenant/handle_provider_deletionapi/src/backend/api/decorators.pyCelery app 与RLSTaskapi/src/backend/config/celery.pyREST/JWT/数据库/Celery 配置参考skills/prowler-api/references/configuration.md生产设置skills/prowler-api/references/production-settings.md建模决策skills/prowler-api/references/modeling-decisions.md测试中央 fixtures / 集成 / 任务api/src/backend/conftest.py、api/src/backend/api/tests/、api/src/backend/tasks/tests/技能文档建议通用 DRF 模式ViewSets、Serializers、Filters、JSON:API参考skills/django-drf/SKILL.mdAPI 测试模式参考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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 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/prowlerProwler 的 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数据库别名用途是否启用 RLSdefaultprowler_user标准 API 查询是adminadmin迁移、鉴权旁路否replicaprowler_user只读查询是admin_replicaadmin管理端只读副本否# 何时使用 admin绕过 RLS from api.db_router import MainRouter User.objects.using(MainRouter.admin_db).get(iduser_id) # 鉴权查询 # 标准查询走 default强制 RLS Provider.objects.filter(connectedTrue) # 需要 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_userRLS 连接的原始定义和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_idset_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 );这里有三个值得注意的设计FORCE ROW LEVEL SECURITY连表属主也受策略约束IS NULL THEN FALSE租户变量未设置时策略直接拒绝所有行——即使有人忘记了 RLS 上下文结果也是看不到任何数据而非看到全部数据这是 fail-closed 设计策略只授予prowler_userAPI 数据库账号指定语句的权限INSERT用WITH CHECK其余语句用USING形成最小权限 行级策略双保险。对全局/共享数据则使用同文件中的BaseSecurityConstraintrls.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_keyTrue, defaultuuid4, editableFalse) name models.CharField(max_length255) inserted_at models.DateTimeField(auto_now_addTrue, editableFalse) updated_at models.DateTimeField(auto_nowTrue, editableFalse) class Meta(RowLevelSecurityProtectedModel.Meta): db_table my_models constraints [ RowLevelSecurityConstraint( fieldtenant_id, namerls_on_%(class)s, statements[SELECT, INSERT, UPDATE, DELETE], ), ] class JSONAPIMeta: resource_name my-models与源码对照rls.py 中RowLevelSecurityProtectedModel是抽象基类继承自models.Model并自带tenant models.ForeignKey(Tenant, on_deletemodels.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, throughResourceTagMapping, # RLS 必需 ) class ResourceTagMapping(RowLevelSecurityProtectedModel): # through 模型必须带 tenant_id 才能启用 RLS resource models.ForeignKey(Resource, on_deletemodels.CASCADE) tag models.ForeignKey(ResourceTag, on_deletemodels.CASCADE) class Meta: constraints [ RowLevelSecurityConstraint( fieldtenant_id, namerls_on_%(class)s, statements[SELECT, INSERT, UPDATE, DELETE], ), ]原因很直接Django 自动生成的 M2M 中间表没有tenant_id列无法为其创建基于租户的策略也就无法阻止跨租户关联。文档同时给出了选型决策树选哪个基类模型租户级数据 →RowLevelSecurityProtectedModel全局/共享数据 →models.ModelBaseSecurityConstraint少见分区时序数据 →PostgresPartitionedModelRowLevelSecurityProtectedModel软删除 → 追加is_deleted字段 ActiveProviderManager。选哪个 Manager常规查询用Model.objects排除已删除需要已删除记录用Model.all_objectsmodels.py 中Provider即定义了all_objects models.Manager()Celery 任务上下文必须先rls_transaction()。选哪个库标准 API 查询走defaultViewSet 自动只读操作走replicaBaseRLSViewSet对 GET 自动鉴权/管理操作走MainRouter.admin_db跨租户查询走admin库谨慎使用。Celery 装饰器顺序shared_task(baseRLSTask, ...)之下先set_tenant设置租户上下文再handle_provider_deletion处理扫描期间被删除的 Provider。异步任务响应模式202 Accepted长耗时操作的标准返回方式是 202 任务引用action(detailTrue, methods[post], url_nameconnection) def connection(self, request, pkNone): with transaction.atomic(): task check_provider_connection_task.delay( provider_idpk, tenant_idself.request.tenant_id ) prowler_task Task.objects.get(idtask.id) serializer TaskSerializer(prowler_task) return Response( dataserializer.data, statusstatus.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()静态方法ProviderUID 格式示例AWS12 位数字123456789012AzureUUID v4a1b2c3d4-e5f6-...GCP6-30 字符小写字母开头my-gcp-projectM365合法域名contoso.onmicrosoft.comKubernetes2-251 字符arn:aws:eks:...GitHub1-39 字符my-orgIaCGit URLhttps://github.com/user/repo.gitOracle CloudOCID 格式ocid1.tenancy.oc1..MongoDB Atlas24 位十六进制507f1f77bcf86cd799439011Alibaba Cloud16 位数字1234567890123456对照 models.py 的ProviderChoices源码当前枚举实际上已扩展到 16 个成员在文档表格基础上还包含 Cloudflare、OpenStack、Image、Google Workspace、Vercel、Okta文档标题写 11 Supported 而表格仅 10 行可视为技能文档滞后于代码的例证。源码中每个校验方法的实现细节也与表格吻合例如validate_aws_uidre.match(r^\d{12}$, value)失败抛出带 JSON:APIpointer/data/attributes/uid的ModelValidationErrormodels.pyvalidate_azure_uid要求是严格 UUID v4 且字符串形式与规范化形式一致models.pyvalidate_gcp_uid6-30 字符、字母开头另兼容domain.com:project-id形式的旧版 App Engine 项目 IDmodels.pyvalidate_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 权限模型文档的权限表权限控制范围MANAGE_USERS用户 CRUD、角色分配MANAGE_ACCOUNT租户设置MANAGE_BILLING计费/订阅MANAGE_PROVIDERSProvider CRUDMANAGE_INTEGRATIONS集成配置MANAGE_SCANS扫描执行UNLIMITED_VISIBILITY可见所有 Provider绕过 provider_groups从 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_idself.request.tenant_id) else: # 按角色分配的 provider_groups 过滤 return Model.objects.filter(provider__inget_providers(user_role))源码印证了两处关键细节get_role(user, tenant_id)permissions.py通过User.roles.using(MainRouter.admin_db).filter(tenant_idtenant_id).first()取第一个角色用户-角色关联表本身在 admin 库中无角色时抛PermissionDenied——这就是技能文档反复强调先get_role()再判断且只返回第一个角色的原因get_providers(role)permissions.py按角色关联的 provider 分组返回去重后的 Provider 查询集角色没有任何分组时返回空集即什么也看不到。视图层则统一用HasPermissions基类permissions.py从视图属性required_permissions读取所需权限列表再对该用户在此租户下的所有角色做任一角色具备该权限的聚合判定——这与get_role()的单角色语义形成互补HasPermissions负责能不能做这个操作get_roleget_providers负责能看到哪些数据。Celery 任务体系队列、装饰器与 Canvas队列划分队列用途scansProwler 扫描执行overview仪表盘聚合严重度、攻击面compliance合规报告生成integrations外部集成Jira、S3、Security HubdeletionProvider/租户删除异步backfill历史数据回填scan-reports输出文件生成CSV、JSON、HTML、PDF按 file-locations.md 的路径表任务定义集中在api/src/backend/tasks/tasks.py业务逻辑分层在tasks/jobs/下scan.pyperform_prowler_scan()、aggregate_findings()、deletion.pydelete_provider()、delete_tenant()、export.pyCSV/JSON/HTML、report.pyPDF 报告、integrations.pyS3/Security Hub/Jira 上传、attack_paths/Neo4j/Cartography 攻击路径等。两个核心装饰器set_tenant的两种模式文档表格模式kwargs 中的tenant_id函数是否收到tenant_idset_tenant默认弹出移除否set_tenant(keep_tenantTrue)读取但保留是从 decorators.py 源码看set_tenant的行为比表格更完整它先用transaction.atomic包住整个任务校验tenant_id是合法 UUID否则ValidationError再通过set_config(api.tenant_id, ...)在当前连接上设置租户变量——与 ViewSet 请求路径用的是同一条机制从而保证请求上下文与任务上下文共享同一套 RLS 语义。handle_provider_deletiondecorators.py处理扫描执行到一半 Provider 被删的竞态捕获ObjectDoesNotExist/DatabaseError/GraphDatabaseQueryException后在rls_transaction内回查 Provider 是否还存在不存在则转抛ProviderDeletedException若任务 kwargs 里只有scan_id会先经 Scan 反查provider_id对图数据库异常还会额外校验租户与 Membership 是否仍存在。任务编写范式shared_task(baseRLSTask, nametask-name, queuescans) set_tenant # 先设置租户上下文 handle_provider_deletion # 后处理被删除的 Provider def my_task(tenant_id: str, provider_id: str): with rls_transaction(tenant_id): provider Provider.objects.get(pkprovider_id)文档推荐的关键任务模式模式说明bindTrue访问self.request.id、self.request.retriesget_task_logger(__name__)Celery 任务中的正确日志方式SoftTimeLimitExceeded捕获后在硬杀前保存进度countdown30延迟 N 秒执行etadatetime(...)指定时间执行配套的安全任务参考实现文档 Quick Reference# 安全的任务入队 —— 事务提交后才入队 with transaction.atomic(): provider Provider.objects.create(**data) transaction.on_commit( lambda: verify_provider_connection.delay( tenant_idstr(request.tenant_id), provider_idstr(provider.id) ) ) # 现代重试模式 shared_task( baseRLSTask, bindTrue, autoretry_for(ConnectionError, TimeoutError, OperationalError), retry_backoffTrue, retry_backoff_max600, retry_jitterTrue, max_retries5, soft_time_limit300, time_limit360, ) set_tenant def sync_provider_data(self, tenant_id, provider_id): with rls_transaction(tenant_id): # ... 任务逻辑 pass # 幂等任务 —— 重试安全 shared_task(baseRLSTask, acks_lateTrue) set_tenant def process_finding(tenant_id, finding_uid, data): with rls_transaction(tenant_id): Finding.objects.update_or_create(uidfinding_uid, defaultsdata)复杂工作流Canvas 原语原语用途chain()顺序执行A → B → Cgroup()并行执行A、B、C 同时进行组合chain 内嵌 group构建复杂工作流注意.si()签名不可变用于阻止结果传递需要用.s()时才传递结果链式/group 示例见 assets/celery_patterns.py。定时任务django-celery-beat操作要点创建调度IntervalSchedule.objects.get_or_create(every24, periodHOURS)创建周期任务使用任务名而非函数kwargsjson.dumps(...)删除周期任务PeriodicTask.objects.filter(name...).delete()避免竞态用countdown5等待数据库提交schedule_provider_scan()的完整示例在 tasks/beat.py 与 assets/celery_patterns.py。Celery 关键配置设置值目的BROKER_VISIBILITY_TIMEOUT8640024h防止长任务被重新入队CELERY_RESULT_BACKENDdjango-db结果存 PostgreSQLCELERY_TASK_TRACK_STARTEDTrue跟踪任务开始soft_time_limit按任务设置抛出SoftTimeLimitExceededtime_limit按任务设置硬杀SIGKILLcelery.py 的源码把这些配置落实得很具体broker 是 Valkey/RedisCELERY_BROKER_URL由VALKEY_*环境变量拼装visibility_timeout默认 86400 秒DJANGO_BROKER_VISIBILITY_TIMEOUTtask_acks_late Truetask_reject_on_worker_lost Trueworker_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__gtestart, id__ltend)为什么用 UUIDv7时间有序的 UUID 让 PostgreSQL 在处理范围查询时可以裁剪prune分区——主键即时间戳id__gte/id__lt的范围条件能直接映射到分区边界。分区参数来自 configuration.mdFINDINGS_TABLE_PARTITION_MONTHS默认 1按月、FINDINGS_TABLE_PARTITION_COUNT默认 7、可选的FINDINGS_TABLE_PARTITION_MAX_AGE_MONTHS过期清理分区管理器实现在api/src/backend/api/partitions.pyPartitionManagerdb_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_size1000) # 带 RLS 的批量创建 create_objects_in_batches(tenant_id, Finding, objects, batch_size500) # 带 RLS 的批量更新 update_objects_in_batches(tenant_id, Finding, objects, fields[status], batch_size500)从 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 导出场景。安全模式汇总文档总结的两张安全速查表租户隔离模式规则ViewSet 中的 RLS经BaseRLSViewSet自动完成——tenant_id 来自 JWTCelery 中的 RLS必须set_tenantrls_transaction(tenant_id)跨租户校验纵深防御验证obj.tenant_id request.tenant_id永不信任用户输入用 JWT 中的request.tenant_id绝不用request.data.get(tenant_id)Admin 库旁路仅用于跨租户管理操作——它会暴露所有租户的数据Celery 任务安全模式规则只用命名任务绝不用来自用户输入的动态任务名校验参数数据库查询前检查 UUID 格式安全入队用transaction.on_commit()在提交后入队现代重试autoretry_for、retry_backoff、retry_jitter时间上限设置soft_time_limit与time_limit防止任务挂死幂等性update_or_create或幂等键完整示例见 assets/security_patterns.py。生产部署检查清单每次生产部署前运行cd api uv run python src/backend/manage.py check --deploy关键设置与风险详见 references/production-settings.md设置生产取值配置错误的风险DEBUGFalse暴露堆栈、设置与 SQLSECRET_KEY环境变量定期轮换会话劫持、CSRF 绕过ALLOWED_HOSTS显式列表Host 头攻击SECURE_SSL_REDIRECTTrue凭据走 HTTPSESSION_COOKIE_SECURETrue会话 Cookie 走 HTTPCSRF_COOKIE_SECURETrueCSRF Token 走 HTTPSECURE_HSTS_SECONDS315360001 年降级攻击CONN_MAX_AGE60或更高连接池耗尽常用命令# 开发 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 --tbshort # 生产检查 cd api uv run python src/backend/manage.py check --deploy文件导航与延伸阅读按 references/file-locations.md 的路径表本文涉及的核心文件在仓库中的位置主题文件RLS 基类模型与约束api/src/backend/api/rls.pyrls_transaction()、批量操作api/src/backend/api/db_utils.py四库路由MainRouterapi/src/backend/api/db_router.pyRBAC 权限api/src/backend/api/rbac/permissions.pyProvider 模型与 UID 校验api/src/backend/api/models.pyset_tenant/handle_provider_deletionapi/src/backend/api/decorators.pyCelery app 与RLSTaskapi/src/backend/config/celery.pyREST/JWT/数据库/Celery 配置参考skills/prowler-api/references/configuration.md生产设置skills/prowler-api/references/production-settings.md建模决策skills/prowler-api/references/modeling-decisions.md测试中央 fixtures / 集成 / 任务api/src/backend/conftest.py、api/src/backend/api/tests/、api/src/backend/tasks/tests/技能文档建议通用 DRF 模式ViewSets、Serializers、Filters、JSON:API参考skills/django-drf/SKILL.mdAPI 测试模式参考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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考