Files
zsglpt/app_config.py
yuyx 2e4b64dcb2 修复37项安全漏洞和Bug
高危修复:
- 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>
2025-12-11 19:35:29 +08:00

231 lines
8.4 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
配置管理模块
集中管理所有配置项,支持环境变量
"""
import os
from datetime import timedelta
from pathlib import Path
# 尝试加载.env文件如果存在
# Bug fix: 添加警告日志,避免静默失败
try:
from dotenv import load_dotenv
env_path = Path(__file__).parent / '.env'
if env_path.exists():
load_dotenv(dotenv_path=env_path)
print(f"✓ 已加载环境变量文件: {env_path}")
except ImportError:
# python-dotenv未安装记录警告
import sys
print("⚠ 警告: python-dotenv未安装将不会加载.env文件。如需使用.env文件请运行: pip install python-dotenv", file=sys.stderr)
# 常量定义
SECRET_KEY_FILE = 'data/secret_key.txt'
def get_secret_key():
"""获取SECRET_KEY优先环境变量"""
# 优先从环境变量读取
secret_key = os.environ.get('SECRET_KEY')
if secret_key:
return secret_key
# 从文件读取
if os.path.exists(SECRET_KEY_FILE):
with open(SECRET_KEY_FILE, 'r') as f:
return f.read().strip()
# 生成新的
new_key = os.urandom(24).hex()
os.makedirs('data', exist_ok=True)
with open(SECRET_KEY_FILE, 'w') as f:
f.write(new_key)
print(f"✓ 已生成新的SECRET_KEY并保存到 {SECRET_KEY_FILE}")
return new_key
class Config:
"""应用配置基类"""
# ==================== Flask核心配置 ====================
SECRET_KEY = get_secret_key()
# ==================== 会话安全配置 ====================
# Bug fix: 生产环境安全警告
SESSION_COOKIE_SECURE = os.environ.get('SESSION_COOKIE_SECURE', 'False').lower() == 'true'
SESSION_COOKIE_HTTPONLY = True # 防止XSS攻击
# SameSite配置HTTP环境使用LaxHTTPS环境使用None
SESSION_COOKIE_SAMESITE = 'None' if os.environ.get('SESSION_COOKIE_SECURE', 'False').lower() == 'true' else 'Lax'
# 自定义cookie名称避免与其他应用冲突
SESSION_COOKIE_NAME = os.environ.get('SESSION_COOKIE_NAME', 'zsglpt_session')
# Cookie路径确保整个应用都能访问
SESSION_COOKIE_PATH = '/'
PERMANENT_SESSION_LIFETIME = timedelta(hours=int(os.environ.get('SESSION_LIFETIME_HOURS', '24')))
# 安全警告检查
@classmethod
def check_security_warnings(cls):
"""检查安全配置,输出警告"""
import sys
warnings = []
env = os.environ.get('FLASK_ENV', 'production')
if env == 'production':
if not cls.SESSION_COOKIE_SECURE:
warnings.append("SESSION_COOKIE_SECURE=False: 生产环境建议启用HTTPS并设置SESSION_COOKIE_SECURE=true")
if warnings:
print("\n⚠ 安全配置警告:", file=sys.stderr)
for w in warnings:
print(f" - {w}", file=sys.stderr)
print("", file=sys.stderr)
# ==================== 数据库配置 ====================
DB_FILE = os.environ.get('DB_FILE', 'data/app_data.db')
DB_POOL_SIZE = int(os.environ.get('DB_POOL_SIZE', '5'))
# ==================== 浏览器配置 ====================
SCREENSHOTS_DIR = os.environ.get('SCREENSHOTS_DIR', '截图')
# ==================== 并发控制配置 ====================
MAX_CONCURRENT_GLOBAL = int(os.environ.get('MAX_CONCURRENT_GLOBAL', '2'))
MAX_CONCURRENT_PER_ACCOUNT = int(os.environ.get('MAX_CONCURRENT_PER_ACCOUNT', '1'))
# ==================== 日志缓存配置 ====================
MAX_LOGS_PER_USER = int(os.environ.get('MAX_LOGS_PER_USER', '100'))
MAX_TOTAL_LOGS = int(os.environ.get('MAX_TOTAL_LOGS', '1000'))
# ==================== 验证码配置 ====================
MAX_CAPTCHA_ATTEMPTS = int(os.environ.get('MAX_CAPTCHA_ATTEMPTS', '5'))
CAPTCHA_EXPIRE_SECONDS = int(os.environ.get('CAPTCHA_EXPIRE_SECONDS', '300'))
# ==================== IP限流配置 ====================
MAX_IP_ATTEMPTS_PER_HOUR = int(os.environ.get('MAX_IP_ATTEMPTS_PER_HOUR', '10'))
IP_LOCK_DURATION = int(os.environ.get('IP_LOCK_DURATION', '3600')) # 秒
# ==================== 超时配置 ====================
PAGE_LOAD_TIMEOUT = int(os.environ.get('PAGE_LOAD_TIMEOUT', '60000')) # 毫秒
DEFAULT_TIMEOUT = int(os.environ.get('DEFAULT_TIMEOUT', '60000')) # 毫秒
# ==================== 知识管理平台配置 ====================
ZSGL_LOGIN_URL = os.environ.get('ZSGL_LOGIN_URL', 'https://postoa.aidunsoft.com/admin/login.aspx')
ZSGL_INDEX_URL_PATTERN = os.environ.get('ZSGL_INDEX_URL_PATTERN', 'index.aspx')
MAX_CONCURRENT_CONTEXTS = int(os.environ.get('MAX_CONCURRENT_CONTEXTS', '100'))
# ==================== 服务器配置 ====================
SERVER_HOST = os.environ.get('SERVER_HOST', '0.0.0.0')
SERVER_PORT = int(os.environ.get('SERVER_PORT', '51233'))
# ==================== SocketIO配置 ====================
SOCKETIO_CORS_ALLOWED_ORIGINS = os.environ.get('SOCKETIO_CORS_ALLOWED_ORIGINS', '*')
# ==================== 日志配置 ====================
LOG_LEVEL = os.environ.get('LOG_LEVEL', 'DEBUG')
LOG_FILE = os.environ.get('LOG_FILE', 'logs/app.log')
LOG_MAX_BYTES = int(os.environ.get('LOG_MAX_BYTES', '10485760')) # 10MB
LOG_BACKUP_COUNT = int(os.environ.get('LOG_BACKUP_COUNT', '5'))
# ==================== 安全配置 ====================
DEBUG = os.environ.get('FLASK_DEBUG', 'False').lower() == 'true'
ALLOWED_SCREENSHOT_EXTENSIONS = {'.png', '.jpg', '.jpeg'}
MAX_SCREENSHOT_SIZE = int(os.environ.get('MAX_SCREENSHOT_SIZE', '10485760')) # 10MB
@classmethod
def validate(cls):
"""验证配置的有效性"""
errors = []
# 验证SECRET_KEY
if not cls.SECRET_KEY or len(cls.SECRET_KEY) < 32:
errors.append("SECRET_KEY长度必须至少32个字符")
# 验证并发配置
if cls.MAX_CONCURRENT_GLOBAL < 1:
errors.append("MAX_CONCURRENT_GLOBAL必须大于0")
if cls.MAX_CONCURRENT_PER_ACCOUNT < 1:
errors.append("MAX_CONCURRENT_PER_ACCOUNT必须大于0")
# 验证数据库配置
if not cls.DB_FILE:
errors.append("DB_FILE不能为空")
if cls.DB_POOL_SIZE < 1:
errors.append("DB_POOL_SIZE必须大于0")
# 验证日志配置
if cls.LOG_LEVEL not in ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']:
errors.append(f"LOG_LEVEL无效: {cls.LOG_LEVEL}")
return errors
@classmethod
def print_config(cls):
"""打印当前配置(隐藏敏感信息)"""
print("=" * 60)
print("应用配置")
print("=" * 60)
print(f"DEBUG模式: {cls.DEBUG}")
print(f"SECRET_KEY: {'*' * 20} (长度: {len(cls.SECRET_KEY)})")
print(f"会话超时: {cls.PERMANENT_SESSION_LIFETIME}")
print(f"Cookie安全: HTTPS={cls.SESSION_COOKIE_SECURE}, HttpOnly={cls.SESSION_COOKIE_HTTPONLY}")
print(f"数据库文件: {cls.DB_FILE}")
print(f"数据库连接池: {cls.DB_POOL_SIZE}")
print(f"并发配置: 全局={cls.MAX_CONCURRENT_GLOBAL}, 单账号={cls.MAX_CONCURRENT_PER_ACCOUNT}")
print(f"日志级别: {cls.LOG_LEVEL}")
print(f"日志文件: {cls.LOG_FILE}")
print(f"截图目录: {cls.SCREENSHOTS_DIR}")
print("=" * 60)
class DevelopmentConfig(Config):
"""开发环境配置"""
DEBUG = True
# 不覆盖SESSION_COOKIE_SECURE使用父类的环境变量配置
class ProductionConfig(Config):
"""生产环境配置"""
DEBUG = False
# 不覆盖SESSION_COOKIE_SECURE使用父类的环境变量配置
# 如需HTTPS请在环境变量中设置 SESSION_COOKIE_SECURE=true
class TestingConfig(Config):
"""测试环境配置"""
DEBUG = True
TESTING = True
DB_FILE = 'data/test_app_data.db'
# 根据环境变量选择配置
config_map = {
'development': DevelopmentConfig,
'production': ProductionConfig,
'testing': TestingConfig,
}
def get_config():
"""获取当前环境的配置"""
env = os.environ.get('FLASK_ENV', 'production')
return config_map.get(env, ProductionConfig)
if __name__ == '__main__':
# 配置验证测试
config = get_config()
errors = config.validate()
if errors:
print("配置验证失败:")
for error in errors:
print(f"{error}")
else:
print("✓ 配置验证通过")
config.print_config()