修复多项安全漏洞和Bug

1. 安全修复:
   - 修复密码重置接口用户枚举漏洞,统一返回消息防止信息泄露
   - 统一密码强度验证为8位以上且包含字母和数字
   - 添加第三方账号密码加密存储(Fernet对称加密)
   - 修复默认管理员弱密码问题,改用随机生成强密码
   - 修复管理员回复XSS漏洞,添加HTML转义
   - 将MD5哈希替换为SHA256

2. 并发Bug修复:
   - 修复日志缓存竞态条件,添加锁保护
   - 修复截图信号量配置变更后不生效问题

3. 其他改进:
   - 添加API参数类型验证和边界检查
   - 新增crypto_utils.py加密工具模块

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2025-12-11 19:14:14 +08:00
parent b9edc4aaa2
commit de51e1b7c7
6 changed files with 293 additions and 70 deletions

View File

@@ -26,6 +26,7 @@ from password_utils import (
verify_password_sha256
)
from app_config import get_config
from crypto_utils import encrypt_password, decrypt_password, migrate_password
# 获取配置
config = get_config()
@@ -473,7 +474,14 @@ def _migrate_to_v5(conn):
# ==================== 管理员相关 ====================
def ensure_default_admin():
"""确保存在默认管理员账号 admin/admin"""
"""确保存在默认管理员账号
安全修复:使用随机生成的强密码代替弱密码'admin'
首次运行时会将密码打印到控制台,请及时修改
"""
import secrets
import string
with db_pool.get_db() as conn:
cursor = conn.cursor()
@@ -482,14 +490,22 @@ def ensure_default_admin():
result = cursor.fetchone()
if result['count'] == 0:
# 创建默认管理员 admin/admin
default_password_hash = hash_password_bcrypt('admin')
# 安全修复生成随机强密码12位包含大小写字母和数字
alphabet = string.ascii_letters + string.digits
random_password = ''.join(secrets.choice(alphabet) for _ in range(12))
default_password_hash = hash_password_bcrypt(random_password)
cursor.execute(
'INSERT INTO admins (username, password_hash) VALUES (?, ?)',
('admin', default_password_hash)
)
conn.commit()
print("✓ 已创建默认管理员账号 (admin/admin)")
print("=" * 60)
print("安全提醒:已创建默认管理员账号")
print(f"用户名: admin")
print(f"密码: {random_password}")
print("请立即登录后修改密码!")
print("=" * 60)
return True
return False
@@ -812,32 +828,45 @@ def delete_user(user_id):
# ==================== 账号相关 ====================
def create_account(user_id, account_id, username, password, remember=True, remark=''):
"""创建账号"""
"""创建账号(密码加密存储)"""
with db_pool.get_db() as conn:
cursor = conn.cursor()
# 安全修复:加密存储第三方账号密码
encrypted_password = encrypt_password(password)
cursor.execute('''
INSERT INTO accounts (id, user_id, username, password, remember, remark)
VALUES (?, ?, ?, ?, ?, ?)
''', (account_id, user_id, username, password, 1 if remember else 0, remark))
''', (account_id, user_id, username, encrypted_password, 1 if remember else 0, remark))
conn.commit()
return cursor.lastrowid
def get_user_accounts(user_id):
"""获取用户的所有账号"""
"""获取用户的所有账号(自动解密密码)"""
with db_pool.get_db() as conn:
cursor = conn.cursor()
cursor.execute('SELECT * FROM accounts WHERE user_id = ? ORDER BY created_at DESC', (user_id,))
return [dict(row) for row in cursor.fetchall()]
accounts = []
for row in cursor.fetchall():
account = dict(row)
# 安全修复:解密第三方账号密码(兼容旧数据)
account['password'] = decrypt_password(account.get('password', ''))
accounts.append(account)
return accounts
def get_account(account_id):
"""获取单个账号"""
"""获取单个账号(自动解密密码)"""
with db_pool.get_db() as conn:
cursor = conn.cursor()
cursor.execute('SELECT * FROM accounts WHERE id = ?', (account_id,))
row = cursor.fetchone()
return dict(row) if row else None
if row:
account = dict(row)
# 安全修复:解密第三方账号密码(兼容旧数据)
account['password'] = decrypt_password(account.get('password', ''))
return account
return None
def update_account_remark(account_id, remark):
@@ -1483,17 +1512,22 @@ def get_feedback_by_id(feedback_id):
def reply_feedback(feedback_id, admin_reply):
"""管理员回复反馈"""
"""管理员回复反馈带XSS防护"""
from app_security import escape_html
with db_pool.get_db() as conn:
cursor = conn.cursor()
cst_tz = pytz.timezone("Asia/Shanghai")
cst_time = datetime.now(cst_tz).strftime("%Y-%m-%d %H:%M:%S")
# 安全修复转义管理员回复防止存储型XSS攻击
safe_reply = escape_html(admin_reply) if admin_reply else ''
cursor.execute('''
UPDATE bug_feedbacks
SET admin_reply = ?, status = 'replied', replied_at = ?
WHERE id = ?
''', (admin_reply, cst_time, feedback_id))
''', (safe_reply, cst_time, feedback_id))
conn.commit()
return cursor.rowcount > 0