ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

Flask多租户SaaS权限管理实战:从数据模型到越权防护

Flask多租户SaaS权限管理实战:从数据模型到越权防护

1. 从单机权限到多租户隔离:这个需求到底难在哪

做过 Flask 单体应用的人,第一次接触多租户 SaaS 的权限管理,大概率会经历一个"想得太简单"的阶段。我当初也是这么想的:不就是给用户表加个tenant_id字段,然后每个查询都带上这个条件吗?真正动手之后才发现,这件事的复杂度远超预期,而且坑几乎全藏在细节里。

先把场景说清楚。所谓多租户 SaaS,就是一套应用实例同时服务多个互相隔离的客户组织(也就是租户)。每个租户有自己的用户、角色、数据,彼此之间绝对不能串。这跟传统的单租户系统有本质区别:传统系统里权限管理只需要回答"这个用户能不能做这件事",而多租户系统要回答两个问题——"这个用户属于哪个租户"以及"在这个租户内他能不能做这件事"。两个问题叠加,权限模型的维度直接翻倍。

为什么很多人第一版都会写崩?因为大家习惯性地把租户隔离当成一个"过滤条件"来处理,而不是当成一个"架构约束"。过滤条件是可以漏的,你写一百个查询,只要有一个忘了加tenant_id,数据就串了。而架构约束是强制的,漏不掉。这个认知差异,决定了你的系统是能上线还是只能跑 Demo。

这篇文章面向的是已经会用 Flask 写基本 CRUD、但对多租户权限还没有系统认知的开发者。我会从数据模型设计、租户上下文注入、RBAC 权限校验、跨租户越权防护这几个层面,把一套可落地的方案拆开讲。中间会穿插我自己踩过的坑,比如租户上下文在异步任务里丢失、超级管理员跨租户操作的边界处理、以及权限缓存导致的数据串号问题。这些是文档里不会写、但线上一定会遇到的东西。

需要提前说明的是,下面涉及的具体表结构、代码组织方式,都是基于常见工程实践给出的参考方案,你可以根据自己的业务规模调整。核心思路比具体实现更重要。

2. 数据模型:租户、用户、角色的三层关系怎么摆

2.1 租户表不是简单的 id + name

很多人设计租户表的时候,随手就写个id和name就完事了。实际跑起来会发现不够用。租户本身也需要状态管理:试用中、正常、欠费停用、已注销。这些状态直接影响用户能不能登录、能不能调用接口。所以租户表至少要包含状态字段和过期时间。

class Tenant(db.Model): __tablename__ = 'tenant' id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(128), nullable=False) code = db.Column(db.String(64), unique=True, nullable=False) # 租户唯一标识 status = db.Column(db.String(16), default='active') # active/suspended/expired expired_at = db.Column(db.DateTime, nullable=True) created_at = db.Column(db.DateTime, default=datetime.utcnow)

这里code字段值得单独说一句。用自增id做租户标识在内部查询没问题,但一旦涉及到对外暴露(比如子域名、API 路径、邀请链接),自增 id 既不安全也不好看。用一个独立的code做业务标识,内部关联仍然用id,这是比较稳妥的做法。

2.2 用户与租户的关系:一对多还是多对多

这是第一个真正需要做决策的地方。一个用户能不能同时属于多个租户?答案取决于你的业务形态。

如果是 B 端工具类产品,一个用户(比如某公司的财务)可能同时是多家公司的员工,那就需要多对多,中间加一张tenant_user关联表,关联表里还可以存这个用户在该租户下的状态(是否已加入、是否被禁用)。

如果是面向单一组织的系统,一个用户只属于一个租户,那直接在用户表加tenant_id就够了。

我个人的建议是:除非业务明确要求,否则第一版用一对多。多对多会带来一系列连锁复杂度——登录时选租户、切换租户、每个租户下角色独立等等。这些复杂度是真实的,不是可以忽略的。等业务真的需要了再迁移,比一开始就过度设计要划算。

一对多的用户表长这样:

class User(db.Model): __tablename__ = 'user' id = db.Column(db.Integer, primary_key=True) tenant_id = db.Column(db.Integer, db.ForeignKey('tenant.id'), nullable=False, index=True) username = db.Column(db.String(64), nullable=False) password_hash = db.Column(db.String(256), nullable=False) status = db.Column(db.String(16), default='active') __table_args__ = ( db.UniqueConstraint('tenant_id', 'username', name='uq_tenant_username'), )

注意那个联合唯一约束。用户名在全局不唯一,但在租户内必须唯一。这意味着 A 租户可以有个叫admin的用户,B 租户也可以有。如果你只给username加唯一索引,第二个租户就建不了同名用户了,这是个很隐蔽的坑。

2.3 角色与权限:RBAC 的经典三张表

权限模型用 RBAC(基于角色的访问控制)是业界共识,没必要自己发明轮子。核心是三张表:角色表、权限表、角色权限关联表,再加上用户角色关联表。

class Role(db.Model): __tablename__ = 'role' id = db.Column(db.Integer, primary_key=True) tenant_id = db.Column(db.Integer, db.ForeignKey('tenant.id'), nullable=False, index=True) name = db.Column(db.String(64), nullable=False) is_system = db.Column(db.Boolean, default=False) # 系统内置角色不可删 class Permission(db.Model): __tablename__ = 'permission' id = db.Column(db.Integer, primary_key=True) code = db.Column(db.String(128), unique=True, nullable=False) # 如 user:create name = db.Column(db.String(128), nullable=False) class RolePermission(db.Model): __tablename__ = 'role_permission' role_id = db.Column(db.Integer, db.ForeignKey('role.id'), primary_key=True) permission_id = db.Column(db.Integer, db.ForeignKey('permission.id'), primary_key=True) class UserRole(db.Model): __tablename__ = 'user_role' user_id = db.Column(db.Integer, db.ForeignKey('user.id'), primary_key=True) role_id = db.Column(db.Integer, db.ForeignKey('role.id'), primary_key=True)

这里有个关键设计点:权限表是全局的,角色表是租户级的。为什么?因为权限点是由代码定义的,比如"创建用户""删除订单"这些能力,全系统就这么多,不会因为租户不同而不同。但角色是租户自己配置的,A 租户可能把"创建用户"和"删除用户"打包成一个"管理员"角色,B 租户可能拆成两个角色。角色归租户,权限归系统,这个边界要划清楚。

2.4 为什么权限点要用字符串 code 而不是数字 id

权限校验的代码里,你肯定不想写if 3 in user_permissions这种魔法数字。用user:create这样的字符串 code,代码可读性直接上一个台阶,而且权限点定义和代码是强绑定的,改起来也直观。代价是字符串比较比整数慢一点点,但权限校验通常有缓存,这点开销可以忽略。

3. 租户上下文:让每个请求都知道"我在为谁服务"

3.1 为什么不能靠每个函数手动传 tenant_id

最原始的做法是每个视图函数、每个 service 方法都接收一个tenant_id参数,然后往下传。写三个接口还行,写到三十个接口你就会发现:总有那么一两个地方忘了传,或者传错了。而且这种手动传递的方式,一旦中间某层用了第三方库(比如某个 ORM 钩子、某个装饰器),参数就断了。

正确的做法是引入租户上下文,类似 Flask 的g对象或者contextvars,让当前请求的租户信息在整个调用链里随时可取,不需要显式传递。

3.2 用 Flask 的 g 对象承载租户信息

Flask 的g对象是请求级别的,天然适合放租户上下文。在请求进入业务逻辑之前,通过一个before_request钩子把租户解析出来塞进g。

from flask import g, request @app.before_request def load_tenant_context(): # 从 JWT、session 或请求头里解析租户标识 tenant_code = resolve_tenant_from_request(request) if not tenant_code: g.tenant = None return tenant = Tenant.query.filter_by(code=tenant_code, status='active').first() if not tenant: abort(403, description='租户不存在或已停用') g.tenant = tenant g.tenant_id = tenant.id

resolve_tenant_from_request具体怎么实现,取决于你的架构。常见的有三种:从 JWT 的 payload 里取、从子域名里取、从请求头X-Tenant-Code里取。JWT 方式最省事,因为登录时就把租户信息签进去了;子域名方式对用户友好,但需要配置泛域名解析;请求头方式最灵活,适合前后端分离。

3.3 租户上下文在异步任务里会丢,这是个大坑

g对象只在请求生命周期内有效。一旦你把任务丢给 Celery 或者后台线程,g.tenant_id就没了。我当初做异步导出功能时就栽在这上面——导出任务跑起来,查数据时g.tenant_id是 None,结果把所有租户的数据都导出来了。幸好是测试环境发现的。

解决办法是:在任务入队时把 tenant_id 作为显式参数传进去,任务函数里再重新构建上下文。

@celery.task def export_data(tenant_id, user_id): # 异步任务里没有 g,需要手动构建 with tenant_context(tenant_id): # 这里的查询会自动带上租户过滤 data = Order.query.all() ...

tenant_context可以是一个上下文管理器,内部用contextvars存当前租户,这样即使不在 Flask 请求里也能工作。这个设计比依赖g更健壮,因为它不绑定 Flask 的请求生命周期。

3.4 一个容易忽略的细节:租户状态校验的时机

租户被停用后,已经登录的用户怎么办?如果只在登录时校验租户状态,那停用之后用户手里的 token 还能继续用,直到过期。这显然不对。

我的做法是在before_request里每次都校验租户状态。听起来每次请求都查一次数据库有点重,但租户信息可以缓存,比如用 Redis 缓存租户状态,TTL 设个 60 秒。这样停用操作最多 60 秒后生效,对绝大多数业务来说足够了。如果对实时性要求极高,可以在停用时主动清除缓存。

4. 权限校验:从装饰器到数据级过滤

4.1 权限装饰器的正确写法

有了租户上下文和 RBAC 表,权限校验就可以做成装饰器,挂在需要保护的接口上。

def require_permission(permission_code): def decorator(f): @wraps(f) def wrapper(*args, **kwargs): user = get_current_user() if not user: abort(401) if not user_has_permission(user, permission_code): abort(403, description='权限不足') return f(*args, **kwargs) return wrapper return decorator @app.route('/api/users', methods=['POST']) @require_permission('user:create') def create_user(): ...

user_has_permission的实现要注意:它需要查用户的所有角色,再查角色对应的所有权限,然后判断目标权限是否在其中。这个查询链比较长,一定要加缓存。

4.2 权限缓存的失效策略

权限数据的特点是读多写极少。一个用户的权限可能几个月都不变,但每个请求都要校验。所以缓存是必须的。

缓存 key 用user_id,value 是这个用户的权限 code 集合。什么时候失效?用户角色变更时、角色权限变更时、用户被禁用时。这里有个坑:如果你只失效单个用户的缓存,那当某个角色的权限变了,所有拥有这个角色的用户缓存都得失效。所以角色权限变更时,要找出所有关联用户,批量清缓存。

def invalidate_role_cache(role_id): user_ids = [ur.user_id for ur in UserRole.query.filter_by(role_id=role_id).all()] for uid in user_ids: redis.delete(f'perm:user:{uid}')

如果用户量很大,这个批量删除可能比较慢。可以考虑给缓存加一个版本号,角色权限变更时递增版本号,缓存 key 里带上版本号,旧版本自然失效。这是更优雅的做法,但实现稍复杂。

4.3 数据级权限:光有接口权限还不够

接口权限解决的是"能不能调用这个接口",但解决不了"能看哪些数据"。比如两个租户的管理员都有order:list权限,但 A 租户的管理员绝对不能看到 B 租户的订单。这就是数据级权限,也叫行级隔离。

实现方式有两种。一种是手动过滤,每个查询都写filter_by(tenant_id=g.tenant_id)。这种方式灵活但容易漏。另一种是自动过滤,通过 SQLAlchemy 的事件钩子或者自定义 Query 类,在查询构造时自动注入租户条件。

from sqlalchemy.orm import Query class TenantQuery(Query): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # 自动注入租户过滤 if hasattr(g, 'tenant_id') and g.tenant_id: # 这里需要判断模型是否有 tenant_id 字段 ...

说实话,自动过滤的实现比较 tricky,因为不是所有模型都有tenant_id(比如全局的权限表就没有)。我的建议是:核心业务表用自动过滤,全局表手动排除。或者更简单粗暴一点,所有业务查询都走一个封装好的tenant_query(Model)函数,强制带上租户条件。约定优于配置,团队里所有人都遵守这个约定,比搞复杂的自动机制更可靠。

4.4 超级管理员的跨租户操作怎么处理

系统通常需要一个超级管理员,能跨租户操作,比如查看所有租户、帮某个租户重置配置。这个角色的权限模型和普通用户不一样,它不受租户隔离约束。

处理方式是在用户表加一个is_super_admin标志,或者在租户表里设一个特殊的系统租户。校验时先判断是不是超管,是的话跳过租户过滤。

但这里要特别小心:超管的操作必须留审计日志。因为超管能跨租户,一旦出问题影响面很大。每次超管跨租户操作,都要记录操作人、目标租户、操作内容、时间。这不是可选项,是必须项。

5. 那些线上才会暴露的坑

5.1 租户 ID 从请求参数里取,等于把门敞开

我见过一些实现,租户标识是从 URL 参数或者请求体里取的,比如/api/orders?tenant_id=5。这是极其危险的。攻击者只要把tenant_id改成别的值,就能访问其他租户的数据。

租户标识必须来自可信来源:JWT 签名、服务端 session、或者经过验证的子域名。绝对不能来自用户可以随意修改的请求参数。这个原则听起来简单,但实际项目中因为图方便而违反的情况并不少见。

5.2 关联查询时的租户泄漏

假设你要查订单,订单关联了用户,用户关联了租户。如果你只对订单表做了租户过滤,但通过order.user.tenant_id去访问用户信息,而用户表没过滤,就可能读到其他租户的用户。这种通过关联对象绕开过滤的情况很隐蔽。

防御方法是:所有涉及租户数据的模型,查询入口都强制带租户条件,不要依赖关联对象的租户字段。如果确实需要跨表查询,用显式的 join 并带上租户条件。

5.3 批量操作的租户校验

批量删除、批量更新这类接口,很容易出问题。比如批量删除订单,接口收到一个 id 列表,如果只校验了"用户有删除权限",但没校验"这些订单都属于当前租户",那用户传一个其他租户的订单 id 进来,就删掉了别人的数据。

正确做法是:批量操作前,先查出这些 id 对应的记录,校验每一条的tenant_id是否等于当前租户。有一条不符就整体拒绝。这个校验不能省。

5.4 权限变更后的 token 问题

用户的权限变了,但他手里的 JWT 还没过期,里面可能还带着旧的权限信息。如果权限信息存在 JWT 里,那变更后必须等 token 过期才生效,这个延迟可能无法接受。

所以我的建议是:JWT 里只放用户 id 和租户 id,不放具体权限。权限每次从缓存或数据库查。这样权限变更后,清掉缓存就立即生效。JWT 只负责身份认证,不负责权限授权,职责分离更清晰。

6. 一套可复用的权限校验流程

把上面的东西串起来,一个完整的请求处理流程是这样的:

  1. 请求进入,before_request钩子解析租户标识,校验租户状态,把tenant_id存入上下文。
  2. 解析用户身份(从 JWT 或 session),校验用户状态,把user_id存入上下文。
  3. 路由匹配后,权限装饰器检查用户是否拥有所需权限(走缓存)。
  4. 业务逻辑执行,所有数据查询自动或手动带上租户过滤。
  5. 涉及写操作时,校验目标数据的租户归属。
  6. 超管操作额外记录审计日志。

这套流程的关键在于每一层都有明确的职责,租户解析、身份认证、权限校验、数据过滤各司其职,不混在一起。这样出问题时也容易定位是哪一层的问题。

最后分享一个我在实际项目中总结的小技巧:写一个租户隔离的测试用例集。针对每个业务接口,都写一个"用 A 租户的身份访问 B 租户数据"的测试,断言返回 403 或 404。这个测试集跑起来很快,但能挡住绝大多数越权问题。每次加新接口,就补一个对应的隔离测试。坚持下来,租户串号这类问题基本不会漏到线上。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进