000e4f8d3a
- auth.py: 新增 privileged_required 装饰器 (admin+manager)，admin_required 仅限用户管理 - 路由权限: fixture/logs/device_logs/test_data 的 admin 检查改为 admin+manager - 前端: 导航栏/删除按钮/配置按钮扩展为 admin+manager 可见 - 用户管理: 角色下拉增加 manager 选项，仍仅 admin 可访问 - 新增 /change-password 路由+模板，所有登录用户可自行修改密码 - edc_server models.py: role COMMENT 更新 + ALTER TABLE 迁移
151 lines
5.2 KiB
Python
151 lines
5.2 KiB
Python
"""认证模块 — Flask-Login 集成"""
|
||
|
||
from flask import Blueprint, render_template, request, redirect, url_for, flash
|
||
from flask_login import LoginManager, login_user, logout_user, login_required, current_user
|
||
from werkzeug.security import check_password_hash
|
||
from app.models import get_user_by_username, insert_log
|
||
|
||
auth_bp = Blueprint("auth", __name__)
|
||
login_manager = LoginManager()
|
||
login_manager.login_view = "auth.login"
|
||
login_manager.login_message = "请先登录"
|
||
|
||
|
||
class User:
|
||
"""Flask-Login 用户对象"""
|
||
def __init__(self, user_dict):
|
||
self.id = user_dict["id"]
|
||
self.username = user_dict["username"]
|
||
self.role = user_dict["role"]
|
||
self.is_active = bool(user_dict.get("is_active", 1))
|
||
|
||
@property
|
||
def is_authenticated(self):
|
||
return True
|
||
|
||
def get_id(self):
|
||
return str(self.id)
|
||
|
||
|
||
@login_manager.user_loader
|
||
def load_user(user_id):
|
||
conn = __import__("app.models", fromlist=["get_conn"]).get_conn()
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute("SELECT * FROM tb_user WHERE id=%s", (int(user_id),))
|
||
row = cur.fetchone()
|
||
finally:
|
||
conn.close()
|
||
return User(row) if row else None
|
||
|
||
|
||
def init_auth(app):
|
||
login_manager.init_app(app)
|
||
|
||
|
||
# ─── 装饰器 ────────────────────────────────────────────────────────
|
||
|
||
def admin_required(f):
|
||
"""要求 admin 角色（仅 admin，manager 不可通过）"""
|
||
from functools import wraps
|
||
@wraps(f)
|
||
@login_required
|
||
def wrapper(*args, **kwargs):
|
||
if current_user.role != "admin":
|
||
return "权限不足", 403
|
||
return f(*args, **kwargs)
|
||
return wrapper
|
||
|
||
|
||
def privileged_required(f):
|
||
"""要求 admin 或 manager 角色"""
|
||
from functools import wraps
|
||
@wraps(f)
|
||
@login_required
|
||
def wrapper(*args, **kwargs):
|
||
if current_user.role not in ("admin", "manager"):
|
||
return "权限不足", 403
|
||
return f(*args, **kwargs)
|
||
return wrapper
|
||
|
||
|
||
# ─── 登录 / 登出 ────────────────────────────────────────────────────
|
||
|
||
@auth_bp.route("/login", methods=["GET", "POST"])
|
||
def login():
|
||
if request.method == "POST":
|
||
username = request.form.get("username", "").strip()
|
||
password = request.form.get("password", "")
|
||
user_dict = get_user_by_username(username)
|
||
ip = request.remote_addr or ""
|
||
|
||
if user_dict and user_dict.get("is_active") and check_password_hash(user_dict["password_hash"], password):
|
||
user = User(user_dict)
|
||
login_user(user)
|
||
insert_log(user.id, user.username, "login", ip=ip, result="ok")
|
||
next_page = request.args.get("next")
|
||
return redirect(next_page or url_for("devices.index"))
|
||
else:
|
||
insert_log(0, username, "login", detail="密码错误或账号禁用", ip=ip, result="error")
|
||
flash("用户名或密码错误")
|
||
return render_template("login.html")
|
||
|
||
|
||
@auth_bp.route("/logout")
|
||
@login_required
|
||
def logout():
|
||
insert_log(current_user.id, current_user.username, "logout",
|
||
ip=request.remote_addr or "", result="ok")
|
||
logout_user()
|
||
return redirect(url_for("auth.login"))
|
||
|
||
|
||
@auth_bp.route("/change-password", methods=["GET", "POST"])
|
||
@login_required
|
||
def change_password():
|
||
"""所有用户自行修改密码"""
|
||
if request.method == "POST":
|
||
old_password = request.form.get("old_password", "")
|
||
new_password = request.form.get("new_password", "").strip()
|
||
confirm_password = request.form.get("confirm_password", "")
|
||
|
||
if not old_password or not new_password:
|
||
flash("所有字段都不能为空")
|
||
return render_template("change_password.html")
|
||
|
||
if len(new_password) < 6:
|
||
flash("新密码至少6位")
|
||
return render_template("change_password.html")
|
||
|
||
if new_password != confirm_password:
|
||
flash("两次输入的新密码不一致")
|
||
return render_template("change_password.html")
|
||
|
||
# 验证旧密码
|
||
from app.models import get_conn, get_user_by_username
|
||
from werkzeug.security import generate_password_hash
|
||
user_dict = get_user_by_username(current_user.username)
|
||
if not user_dict or not check_password_hash(user_dict["password_hash"], old_password):
|
||
flash("原密码错误")
|
||
return render_template("change_password.html")
|
||
|
||
# 更新密码
|
||
conn = get_conn()
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"UPDATE tb_user SET password_hash=%s WHERE id=%s",
|
||
(generate_password_hash(new_password), current_user.id),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
insert_log(current_user.id, current_user.username, "update",
|
||
target="self", detail="修改个人密码",
|
||
result="ok", ip=request.remote_addr or "")
|
||
flash("密码修改成功")
|
||
return redirect(url_for("devices.index"))
|
||
|
||
return render_template("change_password.html")
|