158 lines
5.2 KiB
Python
158 lines
5.2 KiB
Python
"""认证与页面路由：登录/登出/CSRF/权限装饰器/页面（/、/wall）。
|
||
|
||
CSRF 的 before_request 由 web_server 注册（app 级），本模块提供实现。
|
||
"""
|
||
import functools
|
||
import secrets
|
||
|
||
from flask import (Blueprint, jsonify, request, redirect, url_for,
|
||
render_template, make_response, session)
|
||
from flask_login import (login_user, logout_user, login_required, current_user)
|
||
|
||
from core.logger import get_logger
|
||
from core.models import User
|
||
|
||
_log = get_logger("web")
|
||
|
||
bp = Blueprint("auth", __name__)
|
||
|
||
|
||
# ================== CSRF ==================
|
||
def _csrf_token():
|
||
"""获取或生成当前会话的 CSRF token（非 GET 请求需在 X-CSRF-Token 头携带）。"""
|
||
if "csrf_token" not in session:
|
||
session["csrf_token"] = secrets.token_hex(16)
|
||
return session["csrf_token"]
|
||
|
||
|
||
def _csrf_protect():
|
||
"""非安全方法（POST/PUT/DELETE/PATCH）校验 X-CSRF-Token 请求头。"""
|
||
if request.method not in ("POST", "PUT", "DELETE", "PATCH"):
|
||
return
|
||
# 登录表单（未登录，尚无 token）和静态文件跳过
|
||
if request.endpoint in ("login", "static"):
|
||
return
|
||
sess = session.get("csrf_token", "")
|
||
header = request.headers.get("X-CSRF-Token", "")
|
||
if not sess or header != sess:
|
||
return jsonify({"ok": False, "error": "CSRF 校验失败"}), 403
|
||
|
||
|
||
@bp.route("/api/csrf")
|
||
@login_required
|
||
def api_csrf():
|
||
"""获取 CSRF token（前端非 GET 请求需携带 X-CSRF-Token 头）。"""
|
||
return jsonify({"ok": True, "token": _csrf_token()})
|
||
|
||
|
||
# ================== 权限控制 ==================
|
||
PERM_TASKS = "tasks"
|
||
PERM_DEVICES = "devices"
|
||
PERM_APKS = "apks"
|
||
PERM_LOGS = "logs"
|
||
ALL_PERMS = (PERM_TASKS, PERM_DEVICES, PERM_APKS, PERM_LOGS)
|
||
_PERM_LABELS = {PERM_TASKS: "任务管理", PERM_DEVICES: "设备控制",
|
||
PERM_APKS: "应用管理", PERM_LOGS: "日志查看"}
|
||
|
||
|
||
def _has_perm(perm):
|
||
"""当前用户是否拥有指定权限。管理员恒为 True。"""
|
||
u = current_user
|
||
return bool(u and (u.is_admin or u.has_perm(perm)))
|
||
|
||
|
||
def _validate_perms(raw):
|
||
"""校验权限位列表：只保留合法值、去重。返回合法列表。"""
|
||
valid = set(ALL_PERMS)
|
||
out = []
|
||
for p in raw or []:
|
||
if p in valid and p not in out:
|
||
out.append(p)
|
||
return out
|
||
|
||
|
||
def perm_required(perm):
|
||
"""路由装饰器：要求登录且拥有指定业务权限，否则 403。"""
|
||
def deco(fn):
|
||
@functools.wraps(fn)
|
||
@login_required
|
||
def wrapper(*args, **kwargs):
|
||
if not _has_perm(perm):
|
||
return jsonify({"ok": False,
|
||
"error": f"无权限执行此操作（需要权限: {_PERM_LABELS.get(perm, perm)}）"}), 403
|
||
return fn(*args, **kwargs)
|
||
return wrapper
|
||
return deco
|
||
|
||
|
||
def admin_required(fn):
|
||
"""路由装饰器：仅管理员可用（用户管理类接口），否则 403。"""
|
||
@functools.wraps(fn)
|
||
@login_required
|
||
def wrapper(*args, **kwargs):
|
||
if not current_user.is_admin:
|
||
return jsonify({"ok": False, "error": "仅管理员可执行此操作"}), 403
|
||
return fn(*args, **kwargs)
|
||
return wrapper
|
||
|
||
|
||
@bp.route("/api/me")
|
||
@login_required
|
||
def api_me():
|
||
"""当前登录用户信息（含权限位），前端据此隐藏无权限的功能入口。"""
|
||
u = current_user
|
||
return jsonify({"ok": True, "user": {
|
||
"id": u.id,
|
||
"username": u.username,
|
||
"is_admin": bool(u.is_admin),
|
||
# 管理员返回全部权限位，前端统一用 perms 判断
|
||
"perms": list(ALL_PERMS) if u.is_admin else u.get_perms(),
|
||
}})
|
||
|
||
|
||
# ================== 页面路由 ==================
|
||
@bp.route("/")
|
||
@login_required
|
||
def index():
|
||
"""单页应用首页。"""
|
||
resp = make_response(render_template("admin/monitor.html"))
|
||
resp.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
|
||
resp.headers["Pragma"] = "no-cache"
|
||
return resp
|
||
|
||
|
||
@bp.route("/wall")
|
||
@login_required
|
||
def wall():
|
||
"""监控大屏（独立全屏页面，供挂墙/电视展示）。
|
||
|
||
深色控制室风格；设备缩略图按需轮询（每设备 ~5s 一帧），
|
||
20 台设备整体开销约 0.1 核 CPU + 50KB/s，普通电脑无压力。
|
||
"""
|
||
resp = make_response(render_template("admin/wall.html"))
|
||
resp.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
|
||
return resp
|
||
|
||
|
||
@bp.route("/login", methods=["GET", "POST"])
|
||
def login():
|
||
if request.method == "POST":
|
||
username = request.form.get("username", "")
|
||
password = request.form.get("password", "")
|
||
user = User.query.filter_by(username=username).first()
|
||
if user and user.check_password(password):
|
||
login_user(user)
|
||
_csrf_token() # 建立 CSRF token，前端通过 /api/csrf 获取
|
||
_log.info(f"用户 {username} 登录")
|
||
return redirect(request.args.get("next") or url_for("auth.index"))
|
||
return render_template("admin/login.html", error="用户名或密码错误")
|
||
return render_template("admin/login.html", error=None)
|
||
|
||
|
||
@bp.route("/logout")
|
||
@login_required
|
||
def logout():
|
||
_log.info(f"用户 {current_user.username} 登出")
|
||
logout_user()
|
||
return redirect(url_for("auth.login"))
|