3、用户认证与权限管理：RBAC权限模型设计、JWT令牌认证实现、用户注册与登录API、角色与权限的CRUD、中间件权限校验

好，咱们直接进入正题。这一章聊的是ERP系统的「门禁系统」——用户认证与权限管理。说白了，就是解决两个问题：你是谁？你能干什么？

我在做第一个ERP项目时，权限这块吃了大亏。当时图省事，直接在用户表里加了个「is_admin」字段。结果上线第三天，一个普通员工误操作删了整张订单表……嗯，从那以后，我再也不敢轻视权限设计了。

3.1 RBAC权限模型设计

RBAC，全称是Role-Based Access Control，基于角色的访问控制。为什么选它？因为现实世界就是这么运作的。

你想想看，公司里不会直接给张三分配「可以查看财务报表」的权限，而是说「张三你是财务经理，财务经理这个角色自然拥有查看报表的权限」。这就是RBAC的核心思想：用户 ↔ 角色 ↔ 权限。

核心三要素：

  • 用户（User）：系统的操作者，比如张三、李四
  • 角色（Role）：权限的集合，比如财务经理、仓库管理员
  • 权限（Permission）：具体的操作许可，比如「创建订单」、「删除用户」

我个人习惯用五张表来实现：

-- 用户表
CREATE TABLE `sys_user` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `username` varchar(50) NOT NULL,
  `password` varchar(255) NOT NULL,
  `status` tinyint(1) DEFAULT '1',
  `created_at` datetime DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_username` (`username`)
);

-- 角色表
CREATE TABLE `sys_role` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `name` varchar(50) NOT NULL,
  `code` varchar(50) NOT NULL,
  `description` varchar(255) DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_code` (`code`)
);

-- 权限表
CREATE TABLE `sys_permission` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `name` varchar(50) NOT NULL,
  `code` varchar(50) NOT NULL,
  `type` tinyint(1) DEFAULT '1' COMMENT '1:菜单 2:按钮 3:API',
  `parent_id` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`)
);

-- 用户角色关联表
CREATE TABLE `sys_user_role` (
  `user_id` int(11) NOT NULL,
  `role_id` int(11) NOT NULL,
  PRIMARY KEY (`user_id`, `role_id`)
);

-- 角色权限关联表
CREATE TABLE `sys_role_permission` (
  `role_id` int(11) NOT NULL,
  `permission_id` int(11) NOT NULL,
  PRIMARY KEY (`role_id`, `permission_id`)
);

避坑指南：我曾经在权限表里只存了「菜单权限」，结果发现后端API完全没控制。用户虽然看不到菜单，但直接调接口照样能操作。所以权限一定要覆盖到API层面。

3.2 JWT令牌认证实现

JWT，全称JSON Web Token。为什么不用传统的Session？因为ERP系统往往是前后端分离的，Session的跨域问题太头疼了。

JWT的结构很简单：Header.Payload.Signature。说白了就是三段Base64编码的字符串，用点号连起来。

我建议用Python的PyJWT库来实现，代码量很少：

import jwt
from datetime import datetime, timedelta

# 密钥，生产环境一定要用环境变量
SECRET_KEY = "your-secret-key-here"

def create_token(user_id, username, roles):
    """生成JWT令牌"""
    payload = {
        'user_id': user_id,
        'username': username,
        'roles': roles,
        'exp': datetime.utcnow() + timedelta(hours=8),  # 8小时过期
        'iat': datetime.utcnow()  # 签发时间
    }
    token = jwt.encode(payload, SECRET_KEY, algorithm='HS256')
    return token

def verify_token(token):
    """验证JWT令牌"""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
        return payload
    except jwt.ExpiredSignatureError:
        return None  # 令牌过期
    except jwt.InvalidTokenError:
        return None  # 无效令牌

注意：JWT的Payload是Base64编码，不是加密！千万不要把密码等敏感信息放进去。我曾经见过有人把用户手机号直接放JWT里，这相当于把密码写在脸上。

3.3 用户注册与登录API

注册和登录，这是最基础的两个接口。但基础不代表简单，里面坑不少。

先看注册接口：

@app.route('/api/auth/register', methods=['POST'])
def register():
    data = request.get_json()
    username = data.get('username')
    password = data.get('password')
    
    # 参数校验
    if not username or not password:
        return jsonify({'code': 400, 'msg': '用户名和密码不能为空'}), 400
    
    if len(password) < 6:
        return jsonify({'code': 400, 'msg': '密码长度不能少于6位'}), 400
    
    # 检查用户名是否已存在
    if User.query.filter_by(username=username).first():
        return jsonify({'code': 400, 'msg': '用户名已存在'}), 400
    
    # 密码加密存储
    hashed_password = generate_password_hash(password)
    
    # 创建用户，默认分配一个普通角色
    user = User(username=username, password=hashed_password)
    db.session.add(user)
    db.session.flush()  # 获取user.id
    
    # 分配默认角色
    default_role = Role.query.filter_by(code='ROLE_USER').first()
    if default_role:
        user_role = UserRole(user_id=user.id, role_id=default_role.id)
        db.session.add(user_role)
    
    db.session.commit()
    
    return jsonify({'code': 200, 'msg': '注册成功'}), 200

再看登录接口：

@app.route('/api/auth/login', methods=['POST'])
def login():
    data = request.get_json()
    username = data.get('username')
    password = data.get('password')
    
    # 查询用户
    user = User.query.filter_by(username=username).first()
    if not user:
        return jsonify({'code': 401, 'msg': '用户名或密码错误'}), 401
    
    # 验证密码
    if not check_password_hash(user.password, password):
        return jsonify({'code': 401, 'msg': '用户名或密码错误'}), 401
    
    # 检查用户状态
    if user.status == 0:
        return jsonify({'code': 403, 'msg': '账号已被禁用'}), 403
    
    # 获取用户角色
    roles = [ur.role.code for ur in user.roles]
    
    # 生成令牌
    token = create_token(user.id, user.username, roles)
    
    return jsonify({
        'code': 200,
        'data': {
            'token': token,
            'user': {
                'id': user.id,
                'username': user.username
            }
        }
    }), 200

个人经验：登录接口一定要返回「用户名或密码错误」，不要具体说是用户名错了还是密码错了。这是安全常识，防止攻击者通过错误信息猜出有效用户名。

3.4 角色与权限的CRUD

这部分是管理后台的核心功能。说白了就是增删改查，但有几个细节要注意。

我直接给一个角色管理的示例：

@app.route('/api/roles', methods=['GET'])
def get_roles():
    """获取角色列表"""
    roles = Role.query.all()
    return jsonify({
        'code': 200,
        'data': [{
            'id': r.id,
            'name': r.name,
            'code': r.code,
            'description': r.description,
            'permission_count': len(r.permissions)
        } for r in roles]
    })

@app.route('/api/roles', methods=['POST'])
def create_role():
    """创建角色"""
    data = request.get_json()
    
    # 检查角色编码是否唯一
    if Role.query.filter_by(code=data['code']).first():
        return jsonify({'code': 400, 'msg': '角色编码已存在'}), 400
    
    role = Role(
        name=data['name'],
        code=data['code'],
        description=data.get('description', '')
    )
    db.session.add(role)
    db.session.commit()
    
    return jsonify({'code': 200, 'msg': '创建成功'}), 200

@app.route('/api/roles/<int:role_id>/permissions', methods=['PUT'])
def assign_permissions(role_id):
    """给角色分配权限"""
    data = request.get_json()
    permission_ids = data.get('permission_ids', [])
    
    role = Role.query.get(role_id)
    if not role:
        return jsonify({'code': 404, 'msg': '角色不存在'}), 404
    
    # 先清除原有权限，再重新分配
    role.permissions = []
    for pid in permission_ids:
        perm = Permission.query.get(pid)
        if perm:
            role.permissions.append(perm)
    
    db.session.commit()
    
    return jsonify({'code': 200, 'msg': '权限分配成功'}), 200

我曾经踩过的坑：删除角色时忘了检查是否有用户关联。结果删了一个角色，导致一批用户突然失去了所有权限。所以删除角色前一定要先检查用户关联，或者做软删除。

3.5 中间件权限校验

最后一步，也是最关键的一步——在请求到达业务逻辑之前，先拦住它检查权限。

我习惯用装饰器来实现，这样代码最干净：

from functools import wraps

def require_permission(permission_code):
    """权限校验装饰器"""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            # 从请求头获取令牌
            token = request.headers.get('Authorization')
            if not token:
                return jsonify({'code': 401, 'msg': '未登录'}), 401
            
            # 验证令牌
            payload = verify_token(token)
            if not payload:
                return jsonify({'code': 401, 'msg': '令牌无效或已过期'}), 401
            
            # 查询用户权限
            user_id = payload['user_id']
            user = User.query.get(user_id)
            if not user:
                return jsonify({'code': 401, 'msg': '用户不存在'}), 401
            
            # 检查是否有超级管理员角色
            if 'ROLE_ADMIN' in [r.code for r in user.roles]:
                return f(*args, **kwargs)
            
            # 检查具体权限
            user_permissions = set()
            for role in user.roles:
                for perm in role.permissions:
                    user_permissions.add(perm.code)
            
            if permission_code not in user_permissions:
                return jsonify({'code': 403, 'msg': '权限不足'}), 403
            
            return f(*args, **kwargs)
        return decorated_function
    return decorator

# 使用示例
@app.route('/api/orders', methods=['POST'])
@require_permission('order:create')
def create_order():
    # 只有拥有 order:create 权限的用户才能访问
    return jsonify({'code': 200, 'msg': '创建订单成功'})

我建议：权限校验不要只依赖前端路由守卫。前端可以防君子，但防不住直接调API的。后端中间件才是真正的防线。另外，超级管理员角色最好做特殊处理，避免给超级管理员分配所有权限的麻烦。

好了，这一章的内容就到这里。用户认证和权限管理是ERP系统的基石，设计得好，后面开发会非常顺畅。下一章我们聊聊「商品管理模块」，到时候会用到今天讲的权限校验。

← 上一章 📖 返回目录 下一章 →