深度优化: 修复跨服务器登录、密码哈希、仪表盘UI、流量趋势等12项问题

### 严重修复
1. 跨服务器无法登录 - 新增SESSION_COOKIE_DOMAIN/SECURE/NAME配置
2. SOCKS5用户密码明文存储→bcrypt哈希(兼容自动迁移)
3. 仪表盘审计日志不显示→替换为空循环为真实API加载
4. 管理员登录防爆破未生效→auth.py集成check_fail2ban

### 中等问题修复
5. 流量趋势图表无数据→新增record_traffic_snapshot后台线程
6. 网络流量累计值跳变→新增基线差值机制
7. asyncio连接计数非线程安全→加锁保护

### 优化改进
8. API密钥页面截断文本修复
9. 新增gunicorn/bcrypt依赖
10. 更新README环境变量文档和生产建议
11. 登录密码常量时间对比防时序攻击
12. 仪表盘图表添加animation:false防卡顿
This commit is contained in:
Your Name
2026-07-16 22:13:14 +08:00
parent 00d9750662
commit 06470512f6
12 changed files with 186 additions and 27 deletions
+27 -8
View File
@@ -1,13 +1,14 @@
"""登录认证。"""
import os
import hashlib
import secrets
from functools import wraps
from flask import Blueprint, request, session, redirect, url_for
from flask import Blueprint, request, session, redirect, url_for, current_app
bp = Blueprint("auth", __name__)
def _get_admin_creds():
import os
_ENV = os.environ.get("SM_ENV_PATH", os.path.join(os.path.dirname(__file__), ".env"))
pwd = os.environ.get("SM_ADMIN_PASSWORD")
user = os.environ.get("SM_ADMIN_USER", "admin")
@@ -30,12 +31,10 @@ def get_or_set_password():
"""启动时检查密码,如果为空则设置默认值并打印提示。"""
import os
u, p = _get_admin_creds()
if not p:
# 没有密码,设置默认值
if not p or not p.strip():
default_pwd = "admin123"
os.environ["SM_ADMIN_PASSWORD"] = default_pwd
os.environ["SM_ADMIN_USER"] = u
# 写入 .env 文件
_ENV = os.environ.get("SM_ENV_PATH", os.path.join(os.path.dirname(__file__), ".env"))
try:
with open(_ENV, "w") as f:
@@ -49,6 +48,12 @@ def get_or_set_password():
return (u, p)
def get_admin_password_hash():
"""获取管理员密码哈希(用于安全比较)。"""
_, p = _get_admin_creds()
return hashlib.sha256(p.encode("utf-8")).hexdigest()
def login_required(f):
@wraps(f)
def decorated(*args, **kwargs):
@@ -66,12 +71,26 @@ def login():
return redirect(url_for("web.index"))
if request.method == "POST":
u, p = _get_admin_creds()
form_user = request.form.get("username", "")
form_user = request.form.get("username", "").strip()
form_pwd = request.form.get("password", "")
if form_user == u and form_pwd == p:
session["logged_in"] = True
# 防爆破检查
client_ip = request.remote_addr or "unknown"
from services.user_service import UserService
us = UserService()
blocked, remaining = us.check_fail2ban(client_ip)
if blocked:
log.warning("防爆破已阻断来自 %s 的登录尝试", client_ip)
return {"error": "登录过于频繁,请稍后再试"}, 429
# 使用常量时间比较防止时序攻击
if form_user == u and secrets.compare_digest(form_pwd, p):
us.record_auth_attempt(client_ip, success=True)
session.permanent = True
session["logged_in"] = True
session["user"] = u
return redirect(request.args.get("next") or url_for("web.index"))
us.record_auth_attempt(client_ip, success=False)
return {"error": "用户名或密码错误"}, 401
return _render_login()