Flask多租户SaaS权限体系实战：从数据隔离到RBAC落地

Flask多租户SaaS权限体系实战：从数据隔离到RBAC落地

发布时间：2026/10/12 18:25:08
Flask多租户SaaS权限体系实战：从数据隔离到RBAC落地 1. 多租户SaaS权限体系到底难在哪做过单机Web应用的人第一次接触多租户SaaS最容易犯的错就是把它当成给每张表加个tenant_id就完事。我最早接手一个内部工单系统改造成SaaS版本时也是这么想的结果上线两周就被客户投诉数据串了——A公司的管理员在用户列表里看到了B公司的人。那次事故让我彻底明白多租户权限管理不是加字段而是一整套从数据隔离到权限校验的纵深防御体系。这篇文章要聊的就是基于Flask构建多租户SaaS时用户权限管理这一块怎么设计、怎么落地、怎么避坑。核心关键词包括多租户隔离、RBAC权限模型、租户上下文、Flask-SQLAlchemy、JWT鉴权。适合正在做或准备做SaaS产品的后端开发者尤其是从单体应用转型过来的朋友。我会把整套方案的选型逻辑、表结构设计、中间件实现、权限装饰器、常见漏洞都拆开讲代码可以直接抄去改。先说清楚这个体系要解决什么问题。一个典型的多租户SaaS同一套代码服务多个企业客户每个客户有自己的用户、角色、数据。权限管理要同时满足三层第一层是租户隔离保证任何查询都带租户过滤绝不能跨租户访问第二层是租户内RBAC每个租户可以自定义角色和权限A租户的管理员和B租户的管理员是两套独立配置第三层是平台级权限SaaS运营方自己需要超管视角来管理所有租户。这三层叠在一起才是完整的权限模型。很多人一上来就用现成的权限库比如Casbin或者Flask-Principal结果发现多租户场景下要么改不动要么性能拉胯。我的建议是核心逻辑自己写因为多租户的租户上下文注入是框架无关的必须深度嵌入到ORM层和请求生命周期里用第三方库反而增加理解成本。下面我从整体设计开始一层层往下拆。2. 整体架构设计与技术选型思路2.1 为什么选Flask而不是Django这个问题我被问过无数次。Django自带admin、自带auth、自带ORM权限看起来做SaaS更省事。但实际做多租户时Django的auth体系是单租户假设的User模型全局唯一你要改成多租户得动它的核心表升级时痛苦不堪。Flask虽然什么都没有但正因为如此每一层都可以按多租户需求定制。具体来说Flask的**应用上下文App Context和请求上下文Request Context**机制天然适合存放租户信息。你可以在请求进入时解析出租户塞进g对象后续所有ORM查询、权限校验都从g里取租户ID。这种上下文贯穿全链路的能力是Flask做多租户的最大优势。Django要做同样的事得靠中间件加线程局部变量反而更绕。技术栈我推荐这套组合Flask Flask-SQLAlchemy Flask-JWT-Extended PostgreSQL。数据库选PostgreSQL而不是MySQL关键原因是它支持行级安全策略Row Level Security, RLS可以作为租户隔离的最后一道防线。即使应用层代码写漏了租户过滤数据库层还能兜底。这个后面会详细讲。2.2 三种数据隔离方案怎么选多租户数据隔离有三种主流方案选错了后期迁移成本极高必须先想清楚。隔离方案实现方式隔离强度成本适用场景独立数据库每租户一个DB最强最高金融、医疗等强合规共享数据库独立Schema每租户一个Schema强中中大型SaaS共享数据库共享表加tenant_id字段弱最低中小型SaaS、快速起步我个人的经验是起步阶段用共享表加tenant_id但代码层面预留Schema切换能力。为什么因为早期客户少独立Schema的运维成本备份、迁移、连接池会拖慢迭代。但你要在数据访问层做抽象让按租户过滤这件事集中在一个地方将来要升级到独立Schema时只改这一层。共享表方案最大的风险是漏过滤。一条User.query.all()如果忘了加租户条件就是全量泄露。所以我的核心设计原则是永远不要直接使用裸查询所有查询必须经过租户感知的基类或查询方法。这是整套方案的地基。2.3 租户上下文的设计租户上下文是整个体系的枢纽。它要回答一个问题当前这个请求属于哪个租户解析来源通常有三种域名a.example.com、请求头X-Tenant-ID、JWT载荷token里带tenant_id。生产环境我建议以JWT载荷为准域名和请求头作为辅助校验。为什么以JWT为准因为token是登录时签发的租户信息在签发那一刻就绑定了用户无法通过改请求头来越权。如果只信请求头攻击者改个X-Tenant-ID就能访问别的租户数据这是低级但常见的漏洞。域名方式适合做租户识别入口但校验权限时还是要回到token。上下文对象我一般这样设计# context.py from flask import g from flask_jwt_extended import get_jwt class TenantContext: def __init__(self, tenant_id, user_id, roles, is_platform_adminFalse): self.tenant_id tenant_id self.user_id user_id self.roles roles self.is_platform_admin is_platform_admin def set_tenant_context(): claims get_jwt() g.tenant TenantContext( tenant_idclaims.get(tenant_id), user_idclaims.get(sub), rolesclaims.get(roles, []), is_platform_adminclaims.get(is_platform_admin, False) )这个g.tenant会在每个请求开始时被设置后续所有数据访问和权限判断都依赖它。注意is_platform_admin这个标志它决定了是否绕过租户过滤——平台超管需要跨租户查询这是唯一的例外必须显式标记绝不能靠tenant_id为空这种隐式判断。3. 数据库表结构与RBAC模型落地3.1 核心表设计RBAC的经典模型是用户-角色-权限三层多租户下每层都要带tenant_id。我先把表结构列出来再解释每个设计决策。# models.py from flask_sqlalchemy import SQLAlchemy from sqlalchemy import UniqueConstraint, Index db SQLAlchemy() class Tenant(db.Model): __tablename__ tenants id db.Column(db.Integer, primary_keyTrue) name db.Column(db.String(128), nullableFalse) code db.Column(db.String(64), uniqueTrue, nullableFalse) # 租户唯一标识 status db.Column(db.SmallInteger, default1) # 1启用 0禁用 created_at db.Column(db.DateTime, server_defaultdb.func.now()) class User(db.Model): __tablename__ users id db.Column(db.Integer, primary_keyTrue) tenant_id db.Column(db.Integer, db.ForeignKey(tenants.id), nullableFalse) username db.Column(db.String(64), nullableFalse) password_hash db.Column(db.String(256), nullableFalse) email db.Column(db.String(128)) status db.Column(db.SmallInteger, default1) __table_args__ ( UniqueConstraint(tenant_id, username, nameuq_tenant_username), Index(idx_tenant_user, tenant_id, status), ) class Role(db.Model): __tablename__ roles id db.Column(db.Integer, primary_keyTrue) tenant_id db.Column(db.Integer, db.ForeignKey(tenants.id), nullableFalse) name db.Column(db.String(64), nullableFalse) code db.Column(db.String(64), nullableFalse) is_system db.Column(db.Boolean, defaultFalse) # 系统内置角色不可删 __table_args__ ( UniqueConstraint(tenant_id, code, nameuq_tenant_role_code), ) class Permission(db.Model): __tablename__ permissions id db.Column(db.Integer, primary_keyTrue) code db.Column(db.String(128), uniqueTrue, nullableFalse) # 全局权限如 user:create name db.Column(db.String(128), nullableFalse) module db.Column(db.String(64)) # 所属模块便于分组展示 class UserRole(db.Model): __tablename__ user_roles id db.Column(db.Integer, primary_keyTrue) tenant_id db.Column(db.Integer, nullableFalse) user_id db.Column(db.Integer, db.ForeignKey(users.id), nullableFalse) role_id db.Column(db.Integer, db.ForeignKey(roles.id), nullableFalse) __table_args__ ( UniqueConstraint(user_id, role_id, nameuq_user_role), ) class RolePermission(db.Model): __tablename__ role_permissions id db.Column(db.Integer, primary_keyTrue) tenant_id db.Column(db.Integer, nullableFalse) role_id db.Column(db.Integer, db.ForeignKey(roles.id), nullableFalse) permission_id db.Column(db.Integer, db.ForeignKey(permissions.id), nullableFalse) __table_args__ ( UniqueConstraint(role_id, permission_id, nameuq_role_perm), )3.2 几个关键设计决策的解释为什么username的唯一约束是(tenant_id, username)而不是全局唯一因为多租户下A公司有个张三B公司也可以有个张三他们是不同的人。如果全局唯一B公司注册时会被拒绝体验极差。登录时用租户标识用户名来定位用户租户标识可以来自域名或登录表单。为什么Permission表不带tenant_id这是权限的字典表定义系统有哪些权限点比如user:create、order:delete。这些权限点是代码里写死的所有租户共享同一套定义。租户能自定义的是角色即哪些权限点组合成一个角色。这样设计的好处是权限点由开发方统一维护租户只做组合避免租户乱建权限导致校验逻辑失控。为什么UserRole和RolePermission要冗余tenant_id严格来说可以通过role_id关联出来但冗余tenant_id有两个好处一是查询时可以直接用tenant_id过滤避免多一次join二是可以做数据库层的租户隔离策略RLSRLS需要每张表都有tenant_id列。这是为将来上RLS预留的。is_system标志的作用每个租户创建时会自动生成几个内置角色比如超级管理员普通成员。这些角色不允许租户删除或改名否则可能导致租户把自己锁死。is_systemTrue的角色在删除接口里要拦截。3.3 租户感知的查询基类这是整套方案最核心的一环。我见过太多项目在service层到处写filter_by(tenant_idg.tenant.tenant_id)写漏一处就是事故。正确做法是把租户过滤下沉到查询层。# base_query.py from flask import g from sqlalchemy.orm import Query class TenantQuery(Query): def _tenant_filtered(self): # 平台超管不过滤 if getattr(g, tenant, None) and g.tenant.is_platform_admin: return self tenant_id g.tenant.tenant_id # 找到模型上的tenant_id列 model self.column_descriptions[0][entity] if hasattr(model, tenant_id): return self.filter(model.tenant_id tenant_id) return self def all(self): return self._tenant_filtered()._all_original() def first(self): return self._tenant_filtered()._first_original()实际实现时更推荐用SQLAlchemy的with_loader_criteria或者事件监听do_orm_execute这样更优雅# tenant_filter.py from sqlalchemy import event from sqlalchemy.orm import with_loader_criteria from flask import g def register_tenant_filter(db): event.listens_for(db.session, do_orm_execute) def _add_tenant_criteria(execute_state): if not execute_state.is_select: return if not getattr(g, tenant, None): return if g.tenant.is_platform_admin: return tenant_id g.tenant.tenant_id execute_state.statement execute_state.statement.options( with_loader_criteria( TenantMixin, lambda cls: cls.tenant_id tenant_id, include_aliasesTrue ) )这里TenantMixin是所有带租户字段模型的基类。只要模型继承它所有SELECT查询会自动加上tenant_id条件。这个方案的好处是开发者无感知写User.query.all()自动就是当前租户的数据从根本上杜绝漏过滤。注意这个自动过滤只对SELECT生效。INSERT和UPDATE要单独处理INSERT时必须显式设置tenant_idUPDATE时因为查询已经过滤了改不到别的租户数据但保险起见还是建议在模型层加校验。4. 权限校验的完整实现链路4.1 JWT载荷设计登录成功后签发的token载荷里要带够权限判断所需的信息但又不能太大token会随每次请求传输。我的设计是# auth.py from flask_jwt_extended import create_access_token def build_token(user, roles, permissions): claims { sub: user.id, tenant_id: user.tenant_id, username: user.username, roles: [r.code for r in roles], perms: list(permissions), # 权限码集合 is_platform_admin: False } return create_access_token(identityuser.id, additional_claimsclaims)把权限码直接放进token好处是每次权限校验不用查库性能极高。坏处是权限变更后token里的权限不会立即更新需要等token过期或主动刷新。这个权衡怎么定我的经验是权限变更频率低token有效期短比如2小时可以接受。如果业务要求权限实时生效那就token里只放角色权限每次查库并加缓存。权限码的命名规范建议用模块:操作格式比如user:create、user:delete、order:export。这样在装饰器里判断时语义清晰也方便前端做按钮级权限控制。4.2 权限装饰器实现Flask里做权限校验最自然的方式是装饰器。我一般写两个一个校验单个权限一个校验多个权限满足其一或全部。# decorators.py from functools import wraps from flask import g, jsonify from flask_jwt_extended import verify_jwt_in_request def require_permission(*required_perms, modeall): def decorator(fn): wraps(fn) def wrapper(*args, **kwargs): verify_jwt_in_request() if not getattr(g, tenant, None): return jsonify(code401, msg未认证), 401 user_perms set(g.tenant.roles and []) # 从token取 # 实际从g.tenant里取perms user_perms set(getattr(g.tenant, perms, [])) if g.tenant.is_platform_admin: return fn(*args, **kwargs) if mode all: ok all(p in user_perms for p in required_perms) else: ok any(p in user_perms for p in required_perms) if not ok: return jsonify(code403, msg权限不足), 403 return fn(*args, **kwargs) return wrapper return decorator使用起来很直观app.route(/api/users, methods[POST]) require_permission(user:create) def create_user(): ... app.route(/api/orders/int:oid, methods[DELETE]) require_permission(order:delete, order:manage, modeany) def delete_order(oid): ...这里有个细节verify_jwt_in_request()必须在装饰器里调用而不是依赖全局的before_request。因为有些接口是公开的比如登录全局校验会误伤。装饰器方式让权限控制精确到每个路由。4.3 数据级权限不只是功能权限功能权限管的是能不能调这个接口数据级权限管的是能看哪些数据。多租户SaaS里数据级权限至少有两层租户隔离前面讲的自动过滤和租户内的数据范围。租户内数据范围常见的有本人数据、本部门数据、全租户数据。这个用RBAC表达不了需要引入**数据范围Data Scope**概念。我的做法是在角色上加一个data_scope字段class Role(db.Model): # ... 其他字段 data_scope db.Column(db.String(16), defaultself) # self仅本人, dept本部门, all全租户然后在查询时根据当前用户所有角色的最大范围来决定过滤条件。比如用户有销售角色scopeself和销售主管角色scopedept取最大范围dept。这个逻辑要封装成一个函数在需要数据范围过滤的查询里调用。def apply_data_scope(query, model, user): scopes [r.data_scope for r in user.roles] if all in scopes: return query if dept in scopes: return query.filter(model.dept_id user.dept_id) return query.filter(model.owner_id user.id)实操心得数据范围过滤和租户过滤是两个独立的维度不要混在一起写。租户过滤在ORM事件层自动做数据范围过滤在service层显式做。这样职责清晰排查问题时也容易定位是哪一层出的问题。5. 实操过程与关键环节实现5.1 从零搭建的完整步骤假设你已经有一个Flask项目骨架下面是从零接入多租户权限的完整流程。第一步安装依赖并初始化pip install flask flask-sqlalchemy flask-jwt-extended psycopg2-binary passlib第二步定义TenantMixin并注册自动过滤# mixins.py class TenantMixin: declared_attr def tenant_id(cls): return db.Column(db.Integer, nullableFalse, indexTrue)所有需要租户隔离的模型都继承TenantMixin。然后在应用工厂里注册前面写的do_orm_execute事件。第三步实现登录接口签发带租户信息的tokenapp.route(/api/login, methods[POST]) def login(): data request.get_json() tenant_code data[tenant_code] username data[username] password data[password] tenant Tenant.query.filter_by(codetenant_code, status1).first() if not tenant: return jsonify(code401, msg租户不存在), 401 user User.query.filter_by(tenant_idtenant.id, usernameusername).first() if not user or not verify_password(password, user.password_hash): return jsonify(code401, msg用户名或密码错误), 401 roles get_user_roles(user) perms get_role_permissions(roles) token build_token(user, roles, perms) return jsonify(code0, data{token: token})注意登录时User.query会触发自动租户过滤但此时g.tenant还没设置所以过滤不生效——这正好因为登录时我们就是要跨租户查用户。但这也意味着登录接口必须显式带tenant_id条件不能依赖自动过滤。这是个容易踩的坑。第四步请求进入时设置租户上下文app.before_request def load_tenant_context(): # 白名单路径跳过 if request.path in (/api/login, /api/health): return try: verify_jwt_in_request() set_tenant_context() except Exception: return jsonify(code401, msg认证失败), 401第五步初始化租户时创建内置角色def init_tenant_roles(tenant_id): default_roles [ {code: admin, name: 管理员, is_system: True, perms: ALL_PERMS}, {code: member, name: 普通成员, is_system: True, perms: BASIC_PERMS}, ] for r in default_roles: role Role(tenant_idtenant_id, coder[code], namer[name], is_systemTrue) db.session.add(role) db.session.flush() for pcode in r[perms]: perm Permission.query.filter_by(codepcode).first() if perm: db.session.add(RolePermission(tenant_idtenant_id, role_idrole.id, permission_idperm.id)) db.session.commit()5.2 权限缓存的实现每次请求都从token解析权限虽然快但token里权限码多了会撑大token。我的做法是token里只放角色code权限集合用Redis缓存key是perms:{tenant_id}:{user_id}TTL设30分钟。权限变更时主动删缓存。def get_user_perms(user_id, tenant_id): cache_key fperms:{tenant_id}:{user_id} cached redis.get(cache_key) if cached: return set(json.loads(cached)) roles get_user_roles_by_id(user_id, tenant_id) perms set() for r in roles: perms.update(p.code for p in r.permissions) redis.setex(cache_key, 1800, json.dumps(list(perms))) return perms这样token保持精简权限校验走缓存性能和数据新鲜度都兼顾了。缓存失效的时机要覆盖角色权限变更、用户角色变更、角色删除。5.3 平台超管的特殊处理平台超管是SaaS运营方的账号不属于任何租户需要跨租户操作。我的设计是单独一张platform_admins表登录走独立接口签发的token里is_platform_adminTrue且tenant_idNone。app.route(/api/platform/login, methods[POST]) def platform_login(): # 校验平台管理员 admin PlatformAdmin.query.filter_by(username...).first() ... claims {sub: admin.id, is_platform_admin: True, tenant_id: None} token create_access_token(identityadmin.id, additional_claimsclaims) return jsonify(code0, data{token: token})平台超管的接口要单独一组路径前缀/api/platform/并且在这些接口上强制校验is_platform_admin。普通租户token访问平台接口一律拒绝。这个隔离必须在路由层做不能只靠权限码因为权限码可能被误配。6. 常见问题与排查技巧实录6.1 数据串租户的排查思路数据串租户是最严重的问题一旦发生要立即定位。我的排查顺序是确认自动过滤是否生效在do_orm_execute事件里打日志看SQL有没有带tenant_id条件。如果没带检查模型是否继承了TenantMixin或者g.tenant是否为空。检查是否有裸SQLdb.session.execute(SELECT * FROM users)这种原生SQL不会触发ORM事件必须手动加条件。项目里要禁止裸SQL或者封装成强制带租户参数的函数。检查平台超管token是否泄露如果普通用户拿到了超管token所有过滤都会绕过。检查token签发逻辑和密钥管理。检查关联查询join操作如果关联的表没有租户过滤可能通过关联字段泄露数据。确保join的每张表都继承TenantMixin。6.2 常见问题速查表问题现象可能原因排查方法解决方案登录后查不到数据自动过滤生效但tenant_id不匹配打印g.tenant.tenant_id和数据的tenant_id检查token签发时的tenant_id权限校验总是403token里perms为空或缓存未命中解码token看claims检查缓存key和权限初始化跨租户能查到数据模型未继承TenantMixin检查模型定义继承Mixin并重新注册事件平台超管看不到数据is_platform_admin未生效检查claims和过滤逻辑确认超管token的claims正确权限变更不生效缓存未清除检查缓存删除逻辑变更时主动删缓存关联查询泄露数据join表未过滤看生成的SQL确保join表继承Mixin6.3 几个血泪教训教训一不要相信前端传的tenant_id。我见过有项目在接口参数里接收tenant_id然后直接用它查询。攻击者改个参数就能访问别的租户。租户ID只能来自token这是铁律。教训二软删除和多租户的坑。如果用了软删除is_deleted标志查询时要同时过滤is_deletedFalse和tenant_id。这两个条件建议都放进自动过滤里避免遗漏。教训三批量操作的租户校验。批量删除接口如果接收ID列表一定要先校验这些ID是否都属于当前租户再执行删除。不能直接DELETE WHERE id IN (...)否则可能删到别的租户数据。教训四导出功能的权限。导出往往被忽视但它能一次性带走大量数据。导出接口除了功能权限还要加数据范围过滤并且建议加频率限制和审计日志。6.4 审计日志的必要性多租户SaaS里审计日志不是可选项。每个敏感操作创建用户、修改角色、删除数据、导出都要记录谁user_id、在哪个租户tenant_id、做了什么action、对象是什么target、什么时候timestamp、IP是什么。出问题时这是唯一的追溯依据。class AuditLog(db.Model): __tablename__ audit_logs id db.Column(db.BigInteger, primary_keyTrue) tenant_id db.Column(db.Integer, indexTrue) user_id db.Column(db.Integer) action db.Column(db.String(64)) target db.Column(db.String(128)) detail db.Column(db.Text) ip db.Column(db.String(64)) created_at db.Column(db.DateTime, server_defaultdb.func.now(), indexTrue)审计日志表数据量大建议按月分表或定期归档。查询时按tenant_id和created_at建联合索引。7. 性能优化与扩展方向7.1 权限校验的性能瓶颈在哪权限校验本身很快瓶颈通常在权限数据的加载。如果每次请求都查库算权限QPS上不去。前面讲的Redis缓存是标配。再进一步可以把权限判断做成位运算每个权限点对应一个bit位用户的权限集合是一个大整数判断权限就是位与操作。这个方案在权限点固定且数量可控比如几百个时非常高效。PERM_BITS {user:create: 1 0, user:delete: 1 1, ...} def has_perm(user_perm_int, perm_code): return user_perm_int PERM_BITS[perm_code] ! 0位运算方案把权限集合压缩成一个整数缓存和传输都极省空间。缺点是权限点增删要重新分配bit位适合权限点稳定的系统。7.2 分库分表的预留如果将来租户数量增长到几千上万单库压力大就要考虑分库。共享表方案下分库通常按tenant_id哈希。这时候前面设计的租户上下文集中管理就体现出价值了——只需要在数据库连接层根据g.tenant.tenant_id路由到对应库业务代码不用改。def get_db_for_tenant(tenant_id): shard tenant_id % SHARD_COUNT return shard_engines[shard]这个改造的前提是所有查询都经过租户上下文没有裸查询。所以前面强调的自动过滤不只是安全需要也是扩展性的基础。7.3 租户级别的限流多租户SaaS要防止单个租户耗尽资源。限流要按租户维度做而不是按IP或全局。用Redis的滑动窗口或令牌桶key用rate:{tenant_id}:{api}。不同租户可以配置不同的配额这本身也是一种商业化手段——基础版限1000次/天专业版限10000次/天。def check_rate_limit(tenant_id, api, limit): key frate:{tenant_id}:{api}:{int(time.time()) // 60} current redis.incr(key) if current 1: redis.expire(key, 60) if current limit: raise RateLimitExceeded()限流要放在权限校验之前还是之后我的做法是放在认证之后、权限之前。因为限流需要tenant_id而tenant_id来自token所以必须先认证。但限流不应该消耗权限校验的资源所以放在权限之前。8. 写在最后的一些个人体会这套多租户权限体系我在几个项目里迭代过从最早的到处写filter到后来的自动过滤加缓存中间踩的坑基本都写在上面的排查表里了。如果只让我给一条建议那就是把租户隔离做成基础设施而不是业务逻辑。业务代码里不应该出现tenant_id这个字段它应该像空气一样存在但不可见。一旦业务代码里开始出现租户判断就说明抽象漏了早晚出问题。另外权限系统的测试要单独写一套。我一般会写一个测试基类自动创建两个租户每个租户建几个用户和角色然后跑一套跨租户访问的用例确保A租户的用户访问B租户的任何资源都返回403或404。这套测试每次改权限相关代码都要跑是防止回归的最后一道防线。最后分享一个小技巧在开发环境加一个调试接口输入token就能看到解析出的tenant_id、roles、perms排查权限问题时非常省事。生产环境记得关掉。这个接口帮我省了无数次的token解码时间。