企业官网建设流程全解析

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 三种数据隔离方案怎么选

多租户数据隔离有三种主流方案，选错了后期迁移成本极高，必须先想清楚。

我个人的经验是：起步阶段用共享表加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_admin=False): 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_id=claims.get("tenant_id"), user_id=claims.get("sub"), roles=claims.get("roles", []), is_platform_admin=claims.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_key=True) name = db.Column(db.String(128), nullable=False) code = db.Column(db.String(64), unique=True, nullable=False) # 租户唯一标识 status = db.Column(db.SmallInteger, default=1) # 1启用 0禁用 created_at = db.Column(db.DateTime, server_default=db.func.now()) class User(db.Model): __tablename__ = "users" id = db.Column(db.Integer, primary_key=True) tenant_id = db.Column(db.Integer, db.ForeignKey("tenants.id"), nullable=False) username = db.Column(db.String(64), nullable=False) password_hash = db.Column(db.String(256), nullable=False) email = db.Column(db.String(128)) status = db.Column(db.SmallInteger, default=1) __table_args__ = ( UniqueConstraint("tenant_id", "username", name="uq_tenant_username"), Index("idx_tenant_user", "tenant_id", "status"), ) class Role(db.Model): __tablename__ = "roles" id = db.Column(db.Integer, primary_key=True) tenant_id = db.Column(db.Integer, db.ForeignKey("tenants.id"), nullable=False) name = db.Column(db.String(64), nullable=False) code = db.Column(db.String(64), nullable=False) is_system = db.Column(db.Boolean, default=False) # 系统内置角色不可删 __table_args__ = ( UniqueConstraint("tenant_id", "code", name="uq_tenant_role_code"), ) class Permission(db.Model): __tablename__ = "permissions" 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) module = db.Column(db.String(64)) # 所属模块，便于分组展示 class UserRole(db.Model): __tablename__ = "user_roles" id = db.Column(db.Integer, primary_key=True) tenant_id = db.Column(db.Integer, nullable=False) user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False) role_id = db.Column(db.Integer, db.ForeignKey("roles.id"), nullable=False) __table_args__ = ( UniqueConstraint("user_id", "role_id", name="uq_user_role"), ) class RolePermission(db.Model): __tablename__ = "role_permissions" id = db.Column(db.Integer, primary_key=True) tenant_id = db.Column(db.Integer, nullable=False) role_id = db.Column(db.Integer, db.ForeignKey("roles.id"), nullable=False) permission_id = db.Column(db.Integer, db.ForeignKey("permissions.id"), nullable=False) __table_args__ = ( UniqueConstraint("role_id", "permission_id", name="uq_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；二是可以做数据库层的租户隔离策略（RLS），RLS需要每张表都有tenant_id列。这是为将来上RLS预留的。

is_system标志的作用：每个租户创建时会自动生成几个内置角色，比如"超级管理员""普通成员"。这些角色不允许租户删除或改名，否则可能导致租户把自己锁死。is_system=True的角色在删除接口里要拦截。

3.3 租户感知的查询基类

这是整套方案最核心的一环。我见过太多项目在service层到处写filter_by(tenant_id=g.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_aliases=True ) )

这里TenantMixin是所有带租户字段模型的基类。只要模型继承它，所有SELECT查询会自动加上tenant_id条件。这个方案的好处是开发者无感知，写User.query.all()自动就是当前租户的数据，从根本上杜绝漏过滤。

注意：这个自动过滤只对SELECT生效。INSERT和UPDATE要单独处理，INSERT时必须显式设置tenant_id，UPDATE时因为查询已经过滤了，改不到别的租户数据，但保险起见还是建议在模型层加校验。

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(identity=user.id, additional_claims=claims)

把权限码直接放进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, mode="all"): def decorator(fn): @wraps(fn) def wrapper(*args, **kwargs): verify_jwt_in_request() if not getattr(g, "tenant", None): return jsonify(code=401, 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(code=403, 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", mode="any") 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), default="self") # self=仅本人, dept=本部门, all=全租户

然后在查询时根据当前用户所有角色的最大范围来决定过滤条件。比如用户有"销售"角色（scope=self）和"销售主管"角色（scope=dept），取最大范围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, nullable=False, index=True)

所有需要租户隔离的模型都继承TenantMixin。然后在应用工厂里注册前面写的do_orm_execute事件。

第三步：实现登录接口，签发带租户信息的token

@app.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(code=tenant_code, status=1).first() if not tenant: return jsonify(code=401, msg="租户不存在"), 401 user = User.query.filter_by(tenant_id=tenant.id, username=username).first() if not user or not verify_password(password, user.password_hash): return jsonify(code=401, msg="用户名或密码错误"), 401 roles = get_user_roles(user) perms = get_role_permissions(roles) token = build_token(user, roles, perms) return jsonify(code=0, 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(code=401, 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_id=tenant_id, code=r["code"], name=r["name"], is_system=True) db.session.add(role) db.session.flush() for pcode in r["perms"]: perm = Permission.query.filter_by(code=pcode).first() if perm: db.session.add(RolePermission(tenant_id=tenant_id, role_id=role.id, permission_id=perm.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 = f"perms:{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_admin=True且tenant_id=None。

@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(identity=admin.id, additional_claims=claims) return jsonify(code=0, data={"token": token})

平台超管的接口要单独一组，路径前缀/api/platform/，并且在这些接口上强制校验is_platform_admin。普通租户token访问平台接口一律拒绝。这个隔离必须在路由层做，不能只靠权限码，因为权限码可能被误配。

6. 常见问题与排查技巧实录

6.1 数据串租户的排查思路

数据串租户是最严重的问题，一旦发生要立即定位。我的排查顺序是：

1. 确认自动过滤是否生效：在do_orm_execute事件里打日志，看SQL有没有带tenant_id条件。如果没带，检查模型是否继承了TenantMixin，或者g.tenant是否为空。
2. 检查是否有裸SQL：db.session.execute("SELECT * FROM users")这种原生SQL不会触发ORM事件，必须手动加条件。项目里要禁止裸SQL，或者封装成强制带租户参数的函数。
3. 检查平台超管token是否泄露：如果普通用户拿到了超管token，所有过滤都会绕过。检查token签发逻辑和密钥管理。
4. 检查关联查询：join操作如果关联的表没有租户过滤，可能通过关联字段泄露数据。确保join的每张表都继承TenantMixin。

6.2 常见问题速查表

6.3 几个血泪教训

教训一：不要相信前端传的tenant_id。我见过有项目在接口参数里接收tenant_id，然后直接用它查询。攻击者改个参数就能访问别的租户。租户ID只能来自token，这是铁律。

教训二：软删除和多租户的坑。如果用了软删除（is_deleted标志），查询时要同时过滤is_deleted=False和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_key=True) tenant_id = db.Column(db.Integer, index=True) 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_default=db.func.now(), index=True)

审计日志表数据量大，建议按月分表或定期归档。查询时按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 = f"rate:{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解码时间。