237899745/zsglpt
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复37项安全漏洞和Bug

Browse Source 
高危修复:
- app.py: 添加ip_rate_limit_lock线程锁保护IP限流字典
- app.py: 添加validate_ip_port()验证代理IP/端口范围
- database.py: SQL字段名白名单验证防止注入
- playwright_automation.py: 改进浏览器进程强制清理逻辑

中危修复:
- database.py: 统一时区处理函数get_cst_now()
- database.py: 消除循环导入,移动app_security导入到顶部
- playwright_automation.py: 所有bare except改为except Exception
- app_config.py: dotenv导入失败警告+安全配置检查
- db_pool.py: 添加详细异常堆栈日志
- app_security.py: 用户名过滤零宽字符
- database.py: delete_old_task_logs分批删除避免锁表

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
yuyx
2025-12-11 19:35:29 +08:00
parent a25c9fbba0
commit 126d4cbb52
6 changed files with 290 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
database.py
View File

@@ -28,6 +28,25 @@ from password_utils import (
from app_config import get_config
from crypto_utils import encrypt_password, decrypt_password, migrate_password
# Bug fix: 将 app_security 导入移到顶部，避免循环导入
# 注意：如果出现循环导入，需要检查 app_security 是否导入了 database
try:
from app_security import escape_html, sanitize_sql_like_pattern
except ImportError:
# 如果导入失败，提供基础实现
import html
def escape_html(text):
"""基础HTML转义"""
if text is None:
return ''
return html.escape(str(text))
def sanitize_sql_like_pattern(pattern):
"""基础SQL LIKE模式清理"""
if pattern is None:
return ''
return str(pattern).replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_')
# 获取配置
config = get_config()
@@ -37,6 +56,30 @@ DB_FILE = config.DB_FILE
# 数据库版本 (用于迁移管理)
DB_VERSION = 5
# ==================== 时区处理工具函数 ====================
# Bug fix: 统一时区处理，避免混用导致的问题
CST_TZ = pytz.timezone("Asia/Shanghai")
def get_cst_now():
"""获取当前CST时间（统一入口）"""
return datetime.now(CST_TZ)
def get_cst_now_str():
"""获取当前CST时间字符串"""
return get_cst_now().strftime('%Y-%m-%d %H:%M:%S')
def parse_cst_datetime(datetime_str):
"""解析CST时间字符串为带时区的datetime对象
Args:
datetime_str: 格式为 'YYYY-MM-DD HH:MM:SS' 的字符串
Returns:
带CST时区的datetime对象
"""
naive = datetime.strptime(datetime_str, '%Y-%m-%d %H:%M:%S')
return CST_TZ.localize(naive)
def hash_password(password):
"""Password hashing using bcrypt"""
@@ -1040,7 +1083,19 @@ def update_system_config(max_concurrent=None, schedule_enabled=None, schedule_ti
max_concurrent_per_account=None, max_screenshot_concurrent=None, proxy_enabled=None,
proxy_api_url=None, proxy_expire_minutes=None,
auto_approve_enabled=None, auto_approve_hourly_limit=None, auto_approve_vip_days=None):
"""更新系统配置"""
"""更新系统配置
Bug fix: 添加字段名白名单验证，防止SQL注入风险
"""
# 白名单：允许更新的字段名
ALLOWED_FIELDS = {
'max_concurrent_global', 'schedule_enabled', 'schedule_time',
'schedule_browse_type', 'schedule_weekdays', 'max_concurrent_per_account',
'max_screenshot_concurrent', 'proxy_enabled', 'proxy_api_url',
'proxy_expire_minutes', 'auto_approve_enabled', 'auto_approve_hourly_limit',
'auto_approve_vip_days', 'updated_at'
}
with db_pool.get_db() as conn:
cursor = conn.cursor()
updates = []
@@ -1100,6 +1155,11 @@ def update_system_config(max_concurrent=None, schedule_enabled=None, schedule_ti
if updates:
updates.append('updated_at = CURRENT_TIMESTAMP')
# Bug fix: 验证所有字段名都在白名单中
for update_clause in updates:
field_name = update_clause.split('=')[0].strip()
if field_name not in ALLOWED_FIELDS:
raise ValueError(f"非法字段名: {field_name}")
sql = f"UPDATE system_config SET {', '.join(updates)} WHERE id = 1"
cursor.execute(sql, params)
conn.commit()
@@ -1172,8 +1232,7 @@ def get_task_logs(limit=100, offset=0, date_filter=None, status_filter=None,
params.append(user_id_filter)
if account_filter:
# 转义LIKE中的特殊字符，防止绕过过滤
from app_security import sanitize_sql_like_pattern
# 转义LIKE中的特殊字符，防止绕过过滤（使用顶部导入的函数）
safe_filter = sanitize_sql_like_pattern(account_filter)
where_clauses.append("tl.username LIKE ? ESCAPE '\\'")
params.append(f"%{safe_filter}%")
@@ -1266,16 +1325,39 @@ def get_task_stats(date_filter=None):
}
def delete_old_task_logs(days=30):
"""删除N天前的任务日志"""
with db_pool.get_db() as conn:
cursor = conn.cursor()
cursor.execute('''
DELETE FROM task_logs
WHERE created_at < datetime('now', '-' || ? || ' days')
''', (days,))
conn.commit()
return cursor.rowcount
def delete_old_task_logs(days=30, batch_size=1000):
"""删除N天前的任务日志
Bug fix: 分批删除，避免长时间锁表
Args:
days: 删除多少天前的日志
batch_size: 每批删除的数量
Returns:
int: 删除的总行数
"""
total_deleted = 0
while True:
with db_pool.get_db() as conn:
cursor = conn.cursor()
# 分批删除，使用LIMIT避免长时间锁表
cursor.execute('''
DELETE FROM task_logs
WHERE rowid IN (
SELECT rowid FROM task_logs
WHERE created_at < datetime('now', '-' || ? || ' days')
LIMIT ?
)
''', (days, batch_size))
deleted = cursor.rowcount
conn.commit()
if deleted == 0:
break
total_deleted += deleted
return total_deleted
def get_user_run_stats(user_id, date_filter=None):
@@ -1447,9 +1529,7 @@ def clean_old_operation_logs(days=30):
# ==================== Bug反馈管理 ====================
def create_bug_feedback(user_id, username, title, description, contact=''):
"""创建Bug反馈（带XSS防护）"""
from app_security import escape_html
"""创建Bug反馈（带XSS防护）（使用顶部导入的escape_html函数）"""
with db_pool.get_db() as conn:
cursor = conn.cursor()
cst_tz = pytz.timezone("Asia/Shanghai")
@@ -1512,9 +1592,7 @@ def get_feedback_by_id(feedback_id):
def reply_feedback(feedback_id, admin_reply):
"""管理员回复反馈（带XSS防护）"""
from app_security import escape_html
"""管理员回复反馈（带XSS防护）（使用顶部导入的escape_html函数）"""
with db_pool.get_db() as conn:
cursor = conn.cursor()
cst_tz = pytz.timezone("Asia/Shanghai")