160 lines
7.0 KiB
Python
160 lines
7.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""角色与权限（RBAC）。
|
||
|
||
改这个的原因：现在的"权限"是九处散落的 `roles=("admin","operator")` 元组。想给
|
||
运营加一个"能看统计但不能改模型"的角色，或者临时收回某人发版的能力，都得改代
|
||
码、重新发布。角色和能力是运营策略，不该固化在代码里。
|
||
|
||
模型是标准三层：
|
||
|
||
用户 ──role──→ 角色 ──role_permissions──→ 权限码
|
||
|
||
代码里只判**权限码**（`model:write`），永远不判角色名。这样加角色、调能力都只
|
||
是改数据；代码里那句 `require("model:write")` 一个字都不用动。
|
||
|
||
内置的三个角色可以改权限，但不能删——删掉 admin 就没人能进后台了。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
# ── 权限码 ───────────────────────────────────────────────────────────────────
|
||
# 命名规则 `资源:动作`。读写分开，因为"能看不能改"是最常见的诉求。
|
||
PERMISSIONS: list[tuple[str, str, str]] = [
|
||
# (权限码, 中文名, 分组)
|
||
("config:read", "查看模型配置", "配置"),
|
||
("config:write", "修改模型配置", "配置"),
|
||
("model:read", "查看模型清单", "模型"),
|
||
("model:write", "增删改模型与编排", "模型"),
|
||
("review:read", "查看待审队列", "审核"),
|
||
("review:write", "通过/驳回待审回复", "审核"),
|
||
("stats:read", "查看调用统计", "运营"),
|
||
("audit:read", "查看审计日志", "运营"),
|
||
("release:write", "发布客户端版本", "运维"),
|
||
("user:read", "查看用户", "系统"),
|
||
("user:write", "管理用户与角色", "系统"),
|
||
("role:write", "管理角色权限", "系统"),
|
||
("im:read", "查看聊天归档概览", "聊天归档"),
|
||
("im:content:read", "查看聊天正文与素材", "聊天归档"),
|
||
("im:import", "导入聊天记录", "聊天归档"),
|
||
("im:export", "导出聊天记录", "聊天归档"),
|
||
("im:storage:write", "管理素材存储", "聊天归档"),
|
||
("im:identity:write", "管理人员唯一标识", "聊天归档"),
|
||
]
|
||
|
||
PERMISSION_CODES = frozenset(code for code, _name, _group in PERMISSIONS)
|
||
|
||
# ── 内置角色 ─────────────────────────────────────────────────────────────────
|
||
# 这里给的只是**初始**授权。装好之后运营可以在界面上改，改完立刻生效。
|
||
BUILTIN_ROLES: list[tuple[str, str, list[str]]] = [
|
||
("admin", "管理员", sorted(PERMISSION_CODES)),
|
||
(
|
||
"operator",
|
||
"配置员",
|
||
[
|
||
"config:read", "config:write",
|
||
"model:read", "model:write",
|
||
"review:read", "review:write",
|
||
"stats:read", "audit:read",
|
||
"release:write",
|
||
"user:read",
|
||
],
|
||
),
|
||
(
|
||
"viewer",
|
||
"只读用户",
|
||
["config:read", "model:read", "review:read", "stats:read", "user:read"],
|
||
),
|
||
]
|
||
|
||
BUILTIN_ROLE_CODES = frozenset(code for code, _name, _perms in BUILTIN_ROLES)
|
||
|
||
SCHEMA = """
|
||
CREATE TABLE IF NOT EXISTS roles (
|
||
code TEXT PRIMARY KEY,
|
||
name TEXT NOT NULL,
|
||
builtin INTEGER NOT NULL DEFAULT 0,
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL
|
||
);
|
||
CREATE TABLE IF NOT EXISTS permissions (
|
||
code TEXT PRIMARY KEY,
|
||
name TEXT NOT NULL,
|
||
group_name TEXT NOT NULL DEFAULT ''
|
||
);
|
||
CREATE TABLE IF NOT EXISTS role_permissions (
|
||
role_code TEXT NOT NULL REFERENCES roles(code) ON DELETE CASCADE,
|
||
permission_code TEXT NOT NULL REFERENCES permissions(code) ON DELETE CASCADE,
|
||
PRIMARY KEY (role_code, permission_code)
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_role_perms ON role_permissions(role_code);
|
||
"""
|
||
|
||
|
||
def seed(db, now: str) -> None:
|
||
"""建表并写入内置角色。可重复执行。
|
||
|
||
只在角色**第一次**出现时写默认授权——之后运营改过的授权不能被一次重启冲掉。
|
||
"""
|
||
db.executescript(SCHEMA)
|
||
for code, name, group in PERMISSIONS:
|
||
db.execute(
|
||
"""INSERT INTO permissions (code,name,group_name) VALUES (?,?,?)
|
||
ON CONFLICT(code) DO UPDATE SET name=excluded.name,
|
||
group_name=excluded.group_name""",
|
||
(code, name, group),
|
||
)
|
||
# 权限码删掉了就该级联清掉授权，否则界面上会显示一堆不存在的能力
|
||
db.execute(
|
||
"DELETE FROM role_permissions WHERE permission_code NOT IN "
|
||
f"({','.join('?' for _ in PERMISSION_CODES)})",
|
||
tuple(sorted(PERMISSION_CODES)),
|
||
)
|
||
for code, name, defaults in BUILTIN_ROLES:
|
||
existing = db.execute("SELECT code FROM roles WHERE code=?", (code,)).fetchone()
|
||
db.execute(
|
||
"""INSERT INTO roles (code,name,builtin,created_at,updated_at)
|
||
VALUES (?,?,1,?,?)
|
||
ON CONFLICT(code) DO UPDATE SET name=excluded.name, builtin=1,
|
||
updated_at=excluded.updated_at""",
|
||
(code, name, now, now),
|
||
)
|
||
if existing is None:
|
||
db.executemany(
|
||
"INSERT OR IGNORE INTO role_permissions (role_code,permission_code) "
|
||
"VALUES (?,?)",
|
||
[(code, item) for item in defaults],
|
||
)
|
||
|
||
|
||
def permissions_of(db, role_code: str) -> set[str]:
|
||
rows = db.execute(
|
||
"SELECT permission_code FROM role_permissions WHERE role_code=?",
|
||
(str(role_code or ""),),
|
||
).fetchall()
|
||
return {row[0] for row in rows}
|
||
|
||
|
||
def grant(db, role_code: str, codes: list[str], now: str) -> set[str]:
|
||
"""把一个角色的授权整体替换成给定集合。返回实际生效的权限码。
|
||
|
||
整体替换而不是增量：界面上是一组勾选框，用户提交的就是"最终状态"。增量语义
|
||
会让"取消勾选"变成静默无效——那种 bug 只有在出事之后才会被发现。
|
||
"""
|
||
role = db.execute("SELECT code FROM roles WHERE code=?", (role_code,)).fetchone()
|
||
if role is None:
|
||
raise ValueError(f"角色不存在：{role_code}")
|
||
wanted = {str(item).strip() for item in (codes or []) if str(item).strip()}
|
||
unknown = wanted - PERMISSION_CODES
|
||
if unknown:
|
||
raise ValueError(f"未知权限码：{'、'.join(sorted(unknown))}")
|
||
if role_code == "admin" and wanted != set(PERMISSION_CODES):
|
||
# 管理员被削权之后就没人能再把权限加回来了，只能改库。这道闸必须在。
|
||
raise ValueError("管理员角色的权限不可修改，否则会把自己锁在门外")
|
||
db.execute("DELETE FROM role_permissions WHERE role_code=?", (role_code,))
|
||
db.executemany(
|
||
"INSERT INTO role_permissions (role_code,permission_code) VALUES (?,?)",
|
||
[(role_code, item) for item in sorted(wanted)],
|
||
)
|
||
db.execute("UPDATE roles SET updated_at=? WHERE code=?", (now, role_code))
|
||
return wanted
|