阻止多租户应用中的数据泄漏

Source: Dev.to

为什么仅靠应用逻辑不足以保障安全：数据库层面的行级安全（RLS）案例

你已经构建了一个强大的多租户 SaaS。你在每个查询中实现了 tenant_id 过滤器，添加了中间件检查，并编写了单元测试。但应用层的安全是脆弱的：一次遗漏的过滤、一个恶意脚本，或直接的数据库连接都可能绕过你的全部防护。
把安全防线从代码迁移到数据库本身。

简要概述

在本次深度解析中，我们拆除“行级安全（RLS）只是 PostgreSQL 的一个特性”的神话，并展示如何将其无缝集成到你的 SQLAlchemy ORM 与 Alembic 迁移工作流中。我们不仅停留在理论层面，还覆盖了生产环境可直接使用的实现细节。

为什么仅在代码中使用 WHERE tenant_id = ? 是一个定时炸弹

• 过滤器位于应用层，可能被省略或被覆盖。
• 直接访问数据库（例如管理员工具、临时查询）会忽略这些过滤器。
• 一次疏忽就可能泄露所有租户的数据。

步骤化 Alembic 迁移脚本：在不破坏 CI/CD 流程的前提下启用 RLS 并定义策略

1. 创建迁移，为目标表添加 tenant_id 列（如果尚未存在）并在其上启用 RLS。

   # alembic revision script
   from alembic import op
   import sqlalchemy as sa
   
   def upgrade():
       op.execute("ALTER TABLE orders ENABLE ROW LEVEL SECURITY;")
       op.execute("""
           CREATE POLICY tenant_isolation ON orders
           USING (tenant_id = current_setting('app.current_tenant')::uuid);
       """)

2. 添加可逆的降级，删除策略并关闭 RLS。

3. 在正常的 CI/CD 流程中运行迁移；这些语句是幂等的，安全可在生产环境执行。

如何使用 contextvars 与事件监听器向 SQLAlchemy 会话注入动态租户上下文

import contextvars
from sqlalchemy import event
from sqlalchemy.orm import Session

# Context variable that holds the current tenant UUID
current_tenant = contextvars.ContextVar("current_tenant")

def set_tenant(tenant_id: str):
    current_tenant.set(tenant_id)

@event.listens_for(Session, "before_flush")
def apply_tenant_setting(session, flush_context, instances):
    tenant = current_tenant.get(None)
    if tenant:
        session.execute(
            f"SET LOCAL app.current_tenant = '{tenant}'"
        )

• 在每个请求开始时调用 set_tenant()（例如在 FastAPI 中间件里）。
• 所有后续的 ORM 查询会自动继承租户上下文，数据库则会强制执行 RLS 策略。

关键陷阱

• 管理员绕过策略：确保超级用户角色不会自动关闭 RLS；如有需要，为管理员账户创建显式策略。
• 性能索引：为 tenant_id 列（以及 RLS 谓词中使用的任何列）建立索引，以避免全表扫描。
• 策略类型：
  • PERMISSIVE（宽容）策略会向允许的行集合中添加。
  • RESTRICTIVE（限制）策略定义可以访问的精确行集合。根据你的安全模型选择合适的类型。

别再指望开发者永远记得每个过滤器。把安全强加在最关键的地方：数据层。

阅读完整技术指南，获取完整代码示例、迁移模板和测试策略，请访问 https://www.adrianovieira.eng.br/en/posts/architecture/row-level-security-sqlachemy-alembic-guide/。