在 Python 中用 Flask 实现 ABAC

基于角色的访问控制回答的是“编辑能做什么？”很多实际规则光有角色是不够的：作者可以编辑自己的草稿，编辑可以编辑本部门的文章，任何人都可以阅读已发布的文章。这些规则依赖用户和资源的属性，这正是基于属性的访问控制（ABAC）的用途。

本教程用 Flask 搭建一个小型文章 API，并把所有这些规则放在一个 Casbin 策略文件中。同样的模型和策略可以原样用于本教程的 Go 和 Node.js 版本。 Python 中基于角色的规则，见 FastAPI RBAC 教程。如果你还在 RBAC 和 ABAC 之间犹豫，先读选择访问控制模型。

1. 安装

pip install pycasbin flask

2. 编写模型

保存为 model.conf：

[request_definition]
r = sub, obj, act

[policy_definition]
p = sub_rule, obj_type, act

[policy_effect]
e = some(where (p.eft == allow))

[matchers]
m = r.obj.Type == p.obj_type && r.act == p.act && eval(p.sub_rule)

r.sub 和 r.obj 是 Python 对象。匹配器读取它们的属性，例如 r.obj.Type。
每条策略规则都在 sub_rule 中带一个条件。 eval(p.sub_rule) 针对请求对这个条件求值，所以规则放在策略中，而不是模型中。

3. 编写策略

保存为 policy.csv：

p, r.obj.Status == 'published', article, read
p, r.sub.Name == r.obj.Owner, article, read
p, r.sub.Name == r.obj.Owner && r.obj.Status == 'draft', article, update
p, r.sub.Role == 'editor' && r.sub.Department == r.obj.Department, article, update
p, r.sub.Role == 'editor' && r.sub.Department == r.obj.Department, article, publish

每一行可以读作“当条件为真时，允许对文章执行这个操作”：

任何人都可以阅读已发布的文章。
作者可以阅读自己的文章，包括草稿。
作者可以在文章还是草稿时编辑自己的文章。
编辑可以编辑本部门的任何文章。
编辑可以发布本部门的文章。

条件在所有 Casbin 实现中使用相同的语法；PyCasbin 会把 && 和 || 转换为 Python 的 and 和 or。

提示

条件中的字符串字面量用单引号，例如 'published'。字段中间出现双引号会破坏 CSV 解析。如果条件中需要逗号，用双引号把整个字段括起来。

4. 编写应用

ABAC 需要资源本身，而不只是它的 URL，所以检查在文章加载之后进行。 authorize 辅助函数加载文章、询问 Casbin，答案为否时以正确的状态码中止请求。每个视图都先调用它。

app.py
from dataclasses import asdict, dataclass

import casbin
from flask import Flask, abort, request

enforcer = casbin.Enforcer("model.conf", "policy.csv")
app = Flask(__name__)


@dataclass
class User:
    Name: str
    Role: str
    Department: str


@dataclass
class Article:
    Id: str
    Title: str
    Owner: str
    Department: str
    Status: str
    Type: str = "article"


# In a real application these come from your database.
users = {
    "alice": User("alice", "author", "engineering"),
    "bob": User("bob", "editor", "engineering"),
    "carol": User("carol", "editor", "sales"),
    "dave": User("dave", "author", "engineering"),
}

articles = {
    "1": Article("1", "Draft by alice", "alice", "engineering", "draft"),
    "2": Article("2", "Published by alice", "alice", "engineering", "published"),
}


def authorize(article_id: str, action: str) -> Article:
    """Load the article and abort unless the current user may perform the action on it."""
    # Replace this with the user from your session or JWT.
    user = users.get(request.headers.get("X-User", ""))
    if user is None:
        abort(401)
    article = articles.get(article_id)
    if article is None:
        abort(404)
    if not enforcer.enforce(user, article, action):
        abort(403)
    return article


@app.get("/articles/<article_id>")
def read_article(article_id):
    return asdict(authorize(article_id, "read"))


@app.put("/articles/<article_id>")
def update_article(article_id):
    article = authorize(article_id, "update")
    article.Title = (request.get_json(silent=True) or {}).get("title", article.Title)
    return asdict(article)


@app.post("/articles/<article_id>/publish")
def publish_article(article_id):
    article = authorize(article_id, "publish")
    article.Status = "published"
    return asdict(article)

示例用 dataclass 保持简短；任何对象都可以，包括 SQLAlchemy 或 Django 模型，只要策略中用到的属性存在即可。

备注

示例用 X-User 请求头保持简短。 Casbin 只负责授权；实际应用中，用户来自你的认证层，例如 Flask-Login 或经过校验的 JWT。

5. 试一试

flask --app app run --port 5000

文章 1 是 alice 的草稿，文章 2 是她已发布的文章，都属于 engineering 部门。

curl -i -H "X-User: dave"  http://localhost:5000/articles/1                 # 403, someone else's draft
curl -i -H "X-User: dave"  http://localhost:5000/articles/2                 # 200, published
curl -i -X PUT -H "X-User: alice" -H "Content-Type: application/json"      -d '{"title":"New title"}' http://localhost:5000/articles/1            # 200, her own draft
curl -i -X PUT -H "X-User: alice" http://localhost:5000/articles/2          # 403, no longer a draft
curl -i -X PUT -H "X-User: bob"   http://localhost:5000/articles/2          # 200, editor in engineering
curl -i -X PUT -H "X-User: carol" http://localhost:5000/articles/2          # 403, editor in sales
curl -i -X POST -H "X-User: alice" http://localhost:5000/articles/1/publish # 403, authors cannot publish
curl -i -X POST -H "X-User: bob"   http://localhost:5000/articles/1/publish # 200
curl -i -H "X-User: dave"  http://localhost:5000/articles/1                 # 200, now published

6. 在运行时修改规则

条件就是策略中的行，所以可以在服务器运行时添加和删除，并通过 SQLAlchemy 适配器等适配器保存到数据库：

# Admins may edit any article.
enforcer.add_policy("r.sub.Role == 'admin'", "article", "update")

# Which rule allowed this request? Useful for audit logs.
enforcer.enforce_ex(users["bob"], articles["2"], "update")
# (True, ["r.sub.Role == 'editor' && r.sub.Department == r.obj.Department", 'article', 'update'])

ABAC 还是 RBAC？

当权限跟随岗位职责、且不依赖访问的是哪条记录时，使用 RBAC。它更容易审计：“列出编辑能做的所有事情”就是一个简单的查询。见 Python RBAC 教程。
当答案取决于记录本身（所有者、部门、状态或敏感度）时，使用 ABAC。
大多数应用两者结合使用：角色决定用户能使用哪些功能，属性再把范围缩小到他能操作的记录。访问控制模型对比中给出了结合两者的模型。

下一步

完整的 ABAC 参考。 ABAC 详细介绍 JSON 请求参数、引号规则和 eval()。
现成的 Flask 集成。 中间件列表中有 flask-authz，它对每个请求检查路由级规则；上面这类记录级检查仍放在视图中。
多个 worker。 使用 watcher，这样任何一个 Gunicorn 或 uWSGI worker 修改策略后，其他 worker 都会重新加载。
在浏览器中试验。 在线编辑器可以评估带 JSON 属性的 ABAC 请求。

1. 安装​

2. 编写模型​

3. 编写策略​

4. 编写应用​

5. 试一试​

6. 在运行时修改规则​

ABAC 还是 RBAC？​

下一步​