liuwei_x_x/abot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复后台弱密码提示误判并恢复server.py编码

Browse Source 
1. 修复数据库账号存在时仍回退 config.toml 判断，导致每次登录重复提示弱密码的问题。
2. 补齐默认管理员密码从旧配置迁移到数据库的同步逻辑，兼容历史部署。
3. 恢复 server.py 为可读 UTF-8 中文版本，并补充后台登录与弱密码判定的回归测试。
This commit is contained in:
Liu
2026-05-01 10:49:38 +08:00
parent 9a52eb33bf
commit 342b4c0065
4 changed files with 467 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
admin/dashboard/blueprints/auth.py
View File

@@ -1,10 +1,11 @@
from flask import Blueprint, render_template, request, redirect, url_for, session, current_app, jsonify
from functools import wraps
from loguru import logger
import time
from functools import wraps
# 创建认证蓝图
auth_bp = Blueprint('auth', __name__)
from flask import Blueprint, current_app, jsonify, redirect, render_template, request, session, url_for
from loguru import logger
# 创建认证蓝图。
auth_bp = Blueprint("auth", __name__)
def _is_ajax_request() -> bool:
@@ -14,36 +15,38 @@ def _is_ajax_request() -> bool:
return requested_with == "xmlhttprequest" or "application/json" in accept_header
# 登录检查装饰器
def login_required(f):
"""登录检查装饰器。"""
@wraps(f)
def decorated_function(*args, **kwargs):
if not session.get('logged_in'):
return redirect(url_for('auth.login'))
if not session.get("logged_in"):
return redirect(url_for("auth.login"))
server = current_app.dashboard_server
session_timeout_seconds = int(server.get_auth_policy().get("session_timeout_minutes", 480) * 60)
last_activity_at = float(session.get("last_activity_at") or 0.0)
if last_activity_at > 0 and (time.time() - last_activity_at) > session_timeout_seconds:
# 会话超时后统一清理登录态，避免长期闲置会话继续可用。
session.clear()
return redirect(url_for('auth.login'))
return redirect(url_for("auth.login"))
session["last_activity_at"] = time.time()
return f(*args, **kwargs)
return decorated_function
# 登录页面
@auth_bp.route('/login', methods=['GET', 'POST'])
@auth_bp.route("/login", methods=["GET", "POST"])
def login():
"""后台登录页及登录提交入口。"""
error = None
if request.method == 'POST':
# 使用 strip 规避用户误输入首尾空格导致的误判。
username = str(request.form.get('username', '') or '').strip()
password = str(request.form.get('password', '') or '')
if request.method == "POST":
# 用户名做 strip，避免首尾空格导致的误判；密码保留原样，避免改变真实口令。
username = str(request.form.get("username", "") or "").strip()
password = str(request.form.get("password", "") or "")
remote_ip = request.remote_addr or ""
# 从应用上下文获取服务器实例，而不是从蓝图对象
server = current_app.dashboard_server
admin_db = getattr(server, "admin_account_db", None)
lock_status = server.get_login_lock_status(username, remote_ip)
@@ -52,82 +55,89 @@ def login():
error = f"登录失败次数过多，请在 {wait_seconds} 秒后再试"
if _is_ajax_request():
return jsonify({"success": False, "error": error}), 429
return render_template('login.html', error=error)
return render_template("login.html", error=error)
# 优先使用数据库账号体系鉴权；若不可用则回退旧配置模式，保证兼容存量部署。
# 认证边界说明：
# 1. 只要数据库里已经存在该管理员账号，就必须以数据库密码为准；
# 2. 否则用户在后台改密后，config.toml 里的旧密码仍然可能继续生效；
# 3. 仅当数据库不可用，或者该账号尚未迁移进数据库时，才回退到旧配置模式。
login_ok = False
should_use_config_fallback = not bool(admin_db)
if admin_db:
try:
login_ok = admin_db.verify_admin_password(username, password)
admin_row = admin_db.get_admin_by_username(username)
should_use_config_fallback = not bool(admin_row)
if admin_row and int(admin_row.get("status") or 0) == 1:
login_ok = admin_db.verify_password(password, str(admin_row.get("password_hash") or ""))
if login_ok:
admin_db.mark_login_success(username, request.remote_addr or "")
admin_db.mark_login_success(username, remote_ip)
except Exception as e:
logger.error(f"数据库账号登录校验异常，回退配置模式: {e}")
login_ok = False
should_use_config_fallback = True
if not login_ok:
login_ok = (username == server.username and password == server.password)
if not login_ok and should_use_config_fallback:
login_ok = username == server.username and password == server.password
if login_ok:
server.clear_login_failures(username, remote_ip)
session['logged_in'] = True
session["logged_in"] = True
session.permanent = True
session['username'] = username # 存储用户名到session
session['last_activity_at'] = time.time()
# 若账号仍在使用默认/弱口令，则登录后强提示修改密码。
session['force_password_change'] = bool(server.should_force_password_change(username))
session["username"] = username
session["last_activity_at"] = time.time()
# 登录成功后再读取当前账号的安全状态，决定是否强制弹出改密提示。
session["force_password_change"] = bool(server.should_force_password_change(username))
logger.debug(f"Login successful. Session after login: {dict(session)}")
if _is_ajax_request():
return jsonify({"success": True, "redirect_url": url_for('main.index')})
return redirect(url_for('main.index'))
return jsonify({"success": True, "redirect_url": url_for("main.index")})
return redirect(url_for("main.index"))
lock_status = server.mark_login_failure(username, remote_ip)
if lock_status.get("locked"):
wait_seconds = int(lock_status.get("remaining_seconds") or 0)
error = f"登录失败次数过多，请在 {wait_seconds} 秒后再试"
else:
lock_status = server.mark_login_failure(username, remote_ip)
if lock_status.get("locked"):
wait_seconds = int(lock_status.get("remaining_seconds") or 0)
error = f"登录失败次数过多，请在 {wait_seconds} 秒后再试"
else:
remaining = max(0, int(server.get_auth_policy().get("max_failed_attempts", 5)) - int(lock_status.get("failed_count") or 0))
error = '用户名或密码错误'
if remaining > 0:
error = f"{error}，再失败 {remaining} 次将暂时锁定"
if _is_ajax_request():
return jsonify({"success": False, "error": error}), 400
remaining = max(
0,
int(server.get_auth_policy().get("max_failed_attempts", 5)) - int(lock_status.get("failed_count") or 0),
)
error = "用户名或密码错误"
if remaining > 0:
error = f"{error}，再失败 {remaining} 次将暂时锁定"
if _is_ajax_request():
return jsonify({"success": False, "error": error}), 400
return render_template('login.html', error=error)
return render_template("login.html", error=error)
# 登出
@auth_bp.route('/logout')
@auth_bp.route("/logout")
def logout():
"""退出登录。"""
session.clear()
return redirect(url_for('auth.login'))
return redirect(url_for("auth.login"))
@auth_bp.route('/api/auth/security_status', methods=['GET'])
@auth_bp.route("/api/auth/security_status", methods=["GET"])
@login_required
def get_security_status():
"""返回当前登录管理员的安全状态。"""
return jsonify({
"success": True,
"data": {
"force_password_change": bool(session.get("force_password_change", False)),
"session_timeout_minutes": int(current_app.dashboard_server.get_auth_policy().get("session_timeout_minutes", 480)),
return jsonify(
{
"success": True,
"data": {
"force_password_change": bool(session.get("force_password_change", False)),
"session_timeout_minutes": int(
current_app.dashboard_server.get_auth_policy().get("session_timeout_minutes", 480)
),
},
}
})
)
@auth_bp.route('/api/auth/change_password', methods=['POST'])
@auth_bp.route("/api/auth/change_password", methods=["POST"])
@login_required
def change_password():
"""修改当前登录管理员密码。
前端请求参数：
{
"old_password": "旧密码",
"new_password": "新密码",
"confirm_password": "确认新密码"
}
"""
"""修改当前登录管理员密码。"""
server = current_app.dashboard_server
admin_db = getattr(server, "admin_account_db", None)
if not admin_db:
@@ -163,7 +173,7 @@ def change_password():
if not updated:
return jsonify({"success": False, "error": "密码更新失败，请稍后重试"}), 500
# 密码修改成功后，移除“必须修改密码”的会话提示。
# 改密成功后立刻移除“必须修改密码”的会话提示，避免用户当前会话继续被反复打断。
session["force_password_change"] = False
return jsonify({"success": True, "message": "密码修改成功"})
except Exception as e: