RBAC 细粒度权限设计

第 16 章 · RBAC 细粒度权限设计

本章目标：在 api-demo 管理后台引入 RBAC（基于角色的访问控制）；设计 Role、Permission 与 User-Role 多对多 关联表；实现 @permission_required('product:write') 装饰器；补充对象级权限（仅可编辑自己创建的商品）；对照 Django Permission 体系；按角色动态展示 Admin 菜单；编写 admin / editor / viewer 三角色种子数据。

学时建议：5～6 小时（含 2 小时权限联调）

前置：本模块 ch05 ORM、ch06 Flask-Login、ch12 api-demo MVP；ch15 JWT 可选（API 与 Session 可并存）。

16.1 为什么需要 RBAC

ch12 毕业项目的管理页仅区分「已登录 / 未登录」。运营扩大后需细分能力：

┌─────────┐  M:N  ┌─────────┐  M:N  ┌────────────┐
│  User   │◄────►│  Role   │◄────►│ Permission │
└────┬────┘      └─────────┘      └────────────┘
     │ 1:N
     ▼
┌─────────┐  对象级：created_by_id == current_user.id
│ Product │
└─────────┘

api-demo 为教学项目；权限码采用 资源:动作（如 product:write），不映射任何真实业务系统。

16.2 数据模型：多对多关联

api_demo/models/rbac.py：

from api_demo.extensions import db

user_roles = db.Table(
    "user_roles",
    db.Column("user_id", db.Integer, db.ForeignKey("users.id"), primary_key=True),
    db.Column("role_id", db.Integer, db.ForeignKey("roles.id"), primary_key=True),
)
role_permissions = db.Table(
    "role_permissions",
    db.Column("role_id", db.Integer, db.ForeignKey("roles.id"), primary_key=True),
    db.Column("permission_id", db.Integer, db.ForeignKey("permissions.id"), primary_key=True),
)

class Role(db.Model):
    __tablename__ = "roles"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(32), unique=True, nullable=False)
    label = db.Column(db.String(64), nullable=False)
    permissions = db.relationship("Permission", secondary=role_permissions, lazy="selectin")

class Permission(db.Model):
    __tablename__ = "permissions"
    id = db.Column(db.Integer, primary_key=True)
    code = db.Column(db.String(64), unique=True, nullable=False)
    description = db.Column(db.String(128))

扩展 User（api_demo/models/user.py）：

class User(UserMixin, db.Model):
    roles = db.relationship("Role", secondary=user_roles, lazy="selectin")

    def has_permission(self, code: str) -> bool:
        return any(p.code == code for r in self.roles for p in r.permissions)

    def has_role(self, name: str) -> bool:
        return any(r.name == name for r in self.roles)

Product 增加创建者字段（对象级权限依据）：

class Product(db.Model):
    created_by_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
    created_by = db.relationship("User", foreign_keys=[created_by_id])

flask db migrate -m "add rbac and product created_by"
flask db upgrade

16.3 权限码设计规范

约定：全小写 资源:动作；动作用 read / write / delete / manage；避免过细拆分。

16.4 @permission_required 装饰器

api_demo/auth/permissions.py：

from functools import wraps
from flask import abort, jsonify, request
from flask_login import current_user

def permission_required(code: str):
    def decorator(view):
        @wraps(view)
        def wrapped(*args, **kwargs):
            if not current_user.is_authenticated:
                abort(401)
            if not current_user.has_permission(code):
                if request.blueprint and request.blueprint.startswith("api"):
                    return jsonify({"code": 403, "message": "权限不足"}), 403
                abort(403)
            return view(*args, **kwargs)
        return wrapped
    return decorator

使用：

@admin_bp.route("/products/new", methods=["GET", "POST"])
@login_required
@permission_required("product:write")
def product_create():
    ...

Web 与 API 蓝图共用同一装饰器；新路由必须在 Code Review 清单中核对是否已加权限。

16.5 对象级权限：只能编辑自己的商品

功能权限（能否调用编辑接口）与对象权限（能否编辑这一条记录）需分层。

api_demo/auth/object_perms.py：

def can_edit_product(user, product) -> bool:
    if user.has_role("admin"):
        return True
    if user.has_permission("product:write"):
        return product.created_by_id == user.id
    return False

视图组合：

@admin_bp.route("/products/<int:pid>/edit", methods=["GET", "POST"])
@login_required
@permission_required("product:write")
def product_edit(pid):
    product = Product.query.get_or_404(pid)
    if not can_edit_product(current_user, product):
        abort(403)
    ...

列表也需过滤，防 IDOR（篡改 URL 中的 pid 越权）：

def products_query_for_user(user):
    q = Product.query
    if user.has_role("admin"):
        return q
    if user.has_permission("product:read"):
        return q.filter(db.or_(
            Product.created_by_id == user.id,
            Product.is_published.is_(True),
        ))
    return q.filter(False)

16.6 与 Django Permission 对照