211 lines
7.7 KiB
Python
211 lines
7.7 KiB
Python
"""登录与权限（RBAC）。
|
||
|
||
角色 = 一组功能菜单权限（rw 可编辑 / ro 只读 / 没配就是看不到），用户挂一个角色。
|
||
菜单权限同时管两端：
|
||
· 前端按 /auth/me 返回的 menus 过滤侧边栏菜单、隐藏写操作按钮；
|
||
· 后端用 guard 依赖按「请求路径 -> 菜单」做二次校验，直接调接口也绕不过去。
|
||
会话是服务端 token（HttpOnly Cookie），退出登录 / 禁用用户时删 token 立即失效。
|
||
密码只存 pbkdf2 派生值，用标准库，不引额外依赖。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import secrets
|
||
from datetime import datetime, timedelta
|
||
|
||
from fastapi import Depends, HTTPException, Request
|
||
from sqlalchemy.orm import Session
|
||
|
||
from . import config
|
||
from .db import get_db
|
||
from .models import Role, User, UserSession, now
|
||
|
||
SESSION_COOKIE = "unis_session"
|
||
PBKDF2_ROUNDS = 120_000
|
||
|
||
# 功能菜单：键是前后端共用的权限标识，顺序即权限管理页里的展示顺序
|
||
MENUS = [
|
||
("board", "工作看板"),
|
||
("projects", "自研产品"),
|
||
("custom", "定制项目"),
|
||
("focus", "每周工作"),
|
||
("staff", "人力资源"),
|
||
("intake", "更新录入"),
|
||
("settings", "设置"),
|
||
("rbac", "权限管理"),
|
||
]
|
||
MENU_KEYS = [k for k, _ in MENUS]
|
||
MENU_LABELS = dict(MENUS)
|
||
LEVELS = ("ro", "rw")
|
||
|
||
# 接口路径首段 -> 菜单。没列出来的（/meta 等）只要求登录，不看菜单。
|
||
SEGMENT_MENU = {
|
||
"focus": "focus",
|
||
"weekly": "board",
|
||
"board": "board",
|
||
"stats": "board",
|
||
"projects": "projects",
|
||
"categories": "projects",
|
||
"tags": "projects",
|
||
"updates": "projects",
|
||
"tasks": "projects",
|
||
"custom": "custom",
|
||
"staff": "staff",
|
||
"ai": "intake",
|
||
"settings": "settings",
|
||
"rbac": "rbac",
|
||
"periods": "settings",
|
||
}
|
||
# /export/xxx 跟着对应页面走
|
||
EXPORT_MENU = {
|
||
"custom": "custom",
|
||
"custom-updates": "custom",
|
||
"projects": "projects",
|
||
"updates": "projects",
|
||
"staff": "staff",
|
||
"focus": "focus",
|
||
}
|
||
|
||
|
||
def normalize_menus(raw) -> dict:
|
||
"""清洗前端传来的权限表：只留已知菜单 + 合法级别。"""
|
||
out: dict[str, str] = {}
|
||
for key, value in (raw or {}).items():
|
||
if key in MENU_KEYS and value in LEVELS:
|
||
out[key] = value
|
||
return out
|
||
|
||
|
||
def menus_json(menus: dict) -> str:
|
||
return json.dumps(normalize_menus(menus), ensure_ascii=False, sort_keys=True)
|
||
|
||
|
||
# ------------------------------------------------------------------ 密码
|
||
def hash_password(password: str) -> tuple[str, str]:
|
||
salt = secrets.token_hex(16)
|
||
digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), bytes.fromhex(salt), PBKDF2_ROUNDS)
|
||
return digest.hex(), salt
|
||
|
||
|
||
def verify_password(password: str, password_hash: str, salt: str) -> bool:
|
||
try:
|
||
digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), bytes.fromhex(salt), PBKDF2_ROUNDS)
|
||
except ValueError:
|
||
return False
|
||
return hmac.compare_digest(digest.hex(), password_hash)
|
||
|
||
|
||
# ------------------------------------------------------------------ 会话
|
||
def start_session(db: Session, user: User) -> str:
|
||
token = secrets.token_urlsafe(32)
|
||
db.add(UserSession(token=token, user_id=user.id,
|
||
expires_at=now() + timedelta(days=config.SESSION_DAYS)))
|
||
user.last_login_at = now()
|
||
db.commit()
|
||
return token
|
||
|
||
|
||
def revoke_all(db: Session, user_id: int) -> None:
|
||
db.query(UserSession).filter(UserSession.user_id == user_id).delete(synchronize_session=False)
|
||
db.commit()
|
||
|
||
|
||
def user_from_token(db: Session, token: str | None) -> User | None:
|
||
if not token:
|
||
return None
|
||
sess = db.get(UserSession, token)
|
||
if sess is None or sess.expires_at < now():
|
||
return None
|
||
user = db.get(User, sess.user_id)
|
||
if user is None or not user.active:
|
||
return None
|
||
return user
|
||
|
||
|
||
# ------------------------------------------------------------------ 依赖
|
||
READ_METHODS = ("GET", "HEAD", "OPTIONS")
|
||
|
||
|
||
def resolve_rule(path: str, method: str):
|
||
"""请求 -> (菜单, 是否写操作)；菜单为 None 表示只要登录就能访问。"""
|
||
parts = [x for x in path.split("/") if x] # /api/custom/projects -> [...]
|
||
if len(parts) >= 2 and parts[0] == "api" and parts[1] == "export":
|
||
menu = EXPORT_MENU.get(parts[2] if len(parts) > 2 else "")
|
||
elif len(parts) >= 2 and parts[0] == "api":
|
||
menu = SEGMENT_MENU.get(parts[1])
|
||
else:
|
||
menu = None
|
||
if menu is None:
|
||
return None
|
||
# 周期表：所有页面顶栏都要读，只有新建/删除算设置页的写操作
|
||
if len(parts) >= 2 and parts[1] == "periods" and method in READ_METHODS:
|
||
return None
|
||
return menu, method not in READ_METHODS
|
||
|
||
|
||
def ensure_seed(db: Session) -> list[str]:
|
||
"""首次启动（或升级后）补内置角色与管理员账号，返回新建的提示项。"""
|
||
notes: list[str] = []
|
||
presets = [
|
||
("管理员", "全部菜单可编辑，含设置与权限管理",
|
||
{k: "rw" for k in MENU_KEYS}, True),
|
||
("部门成员", "业务菜单可编辑，不能改设置与权限",
|
||
{k: "rw" for k in MENU_KEYS if k not in ("settings", "rbac")}, True),
|
||
("只读访客", "只能查看，不能修改任何数据",
|
||
{k: "ro" for k in MENU_KEYS if k != "rbac"}, True),
|
||
]
|
||
role_map: dict[str, Role] = {r.name: r for r in db.query(Role).all()}
|
||
for name, desc, menus, builtin in presets:
|
||
role = role_map.get(name)
|
||
if role is None:
|
||
role = Role(name=name, description=desc, menus=menus_json(menus), is_builtin=builtin)
|
||
db.add(role)
|
||
db.flush()
|
||
role_map[name] = role
|
||
notes.append(f"已创建内置角色「{name}」")
|
||
if not db.query(User).count():
|
||
pw_hash, salt = hash_password(config.ADMIN_PASSWORD)
|
||
db.add(User(name=config.ADMIN_USERNAME, username=config.ADMIN_USERNAME,
|
||
password_hash=pw_hash, password_salt=salt,
|
||
role_id=role_map["管理员"].id, active=True))
|
||
db.commit()
|
||
notes.append(f"已创建管理员账号 {config.ADMIN_USERNAME}（请尽快在用户设置里改密码）")
|
||
db.commit()
|
||
return notes
|
||
|
||
|
||
def guard(request: Request, db: Session = Depends(get_db)) -> User | None:
|
||
"""挂在 /api 子应用上的总闸：先验登录，再按菜单权限放行。"""
|
||
path = request.url.path
|
||
if path.rstrip("/") == "/api/auth/login":
|
||
return None
|
||
user = user_from_token(db, request.cookies.get(SESSION_COOKIE))
|
||
if user is None:
|
||
raise HTTPException(status_code=401, detail="登录已失效，请重新登录")
|
||
if path.startswith("/api/rbac/"):
|
||
# 权限管理自身必须显式授权，普通用户即使能读别的菜单也不能碰
|
||
if "rbac" not in user.menus:
|
||
raise HTTPException(status_code=403, detail="没有「权限管理」的访问权限")
|
||
if request.method not in READ_METHODS and user.menus.get("rbac") != "rw":
|
||
raise HTTPException(status_code=403, detail="「权限管理」是只读权限，不能修改")
|
||
return user
|
||
rule = resolve_rule(path, request.method)
|
||
if rule is None:
|
||
return user
|
||
menu, need_write = rule
|
||
level = user.menus.get(menu)
|
||
if not level:
|
||
raise HTTPException(status_code=403, detail=f"没有「{MENU_LABELS.get(menu, menu)}」的访问权限")
|
||
if need_write and level != "rw":
|
||
raise HTTPException(status_code=403, detail=f"「{MENU_LABELS.get(menu, menu)}」是只读权限，不能修改")
|
||
return user
|
||
|
||
|
||
def current_user(request: Request, db: Session = Depends(get_db)) -> User:
|
||
user = user_from_token(db, request.cookies.get(SESSION_COOKIE))
|
||
if user is None:
|
||
raise HTTPException(status_code=401, detail="请先登录")
|
||
return user
|