auto_control/web/auth.py

"""认证与页面路由：登录/登出/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} 登录")
            try:
                from core import notifier
                notifier.notify("user.login", username=username)
            except Exception:
                pass
            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"))