fix:批量修复了一些问题

2026-09-19 22:51:35 +08:00
parent 835d7217d0
commit 8e6ad5394e
44 changed files with 4921 additions and 395 deletions
@@ -0,0 +1,121 @@
+"""生产环境启动安全校验:缺失关键配置则拒绝启动(生产)或警告(开发)。
+
+校验项:
+  - SECRET_KEY 非空且非弱默认值
+  - 鉴权已配置(密码哈希 / .env 明文密码 / API Key 任一)
+  - DATABASE_URL 不使用示例弱密码(football:football)
+
+与 deps.py 的 fail-closed 互补:此处是「启动时一次性校验 + 明确报错」,
+避免生产带着危险配置上线却只在被攻击时才暴露。
+"""
+from __future__ import annotations
+
+import logging
+import sys
+
+from src.core.config import settings
+from src.core.runtime_config import get_admin_password_hash
+
+logger = logging.getLogger(__name__)
+
+# 明显的弱 SECRET_KEY 黑名单(大小写无关)
+_WEAK_SECRET_KEYS = {
+    "", "changeme", "secret", "password", "123456", "admin",
+    "default", "dev", "development", "test", "example",
+    "openssl rand -base64 32",  # 有人把生成指令直接粘进去
+}
+_MIN_SECRET_KEY_LEN = 16
+
+# 示例弱数据库密码(仅识别最明显的;自定义强密码不受影响)
+_WEAK_DB_PATTERNS = ("football:football@", "admin:admin@", "password@", "123456@")
+
+
+class SecurityCheckError(Exception):
+    """生产环境安全校验失败。"""
+
+
+async def _auth_configured() -> bool:
+    """运行时鉴权是否已配置(含数据库密码哈希/.env 明文/API Key)。"""
+    if await get_admin_password_hash():
+        return True
+    if settings.ADMIN_PASSWORD or settings.ADMIN_API_KEY:
+        return True
+    return False
+
+
+def _check_secret_key() -> list[str]:
+    """返回 SECRET_KEY 的问题列表(空=通过)。"""
+    problems: list[str] = []
+    key = settings.SECRET_KEY
+    if not key:
+        problems.append("SECRET_KEY 未设置,加密与会话签名无法保障")
+        return problems
+    if key.lower().strip() in _WEAK_SECRET_KEYS:
+        problems.append(f"SECRET_KEY 为弱默认值({key[:20]}...),请生成强随机值: openssl rand -base64 32")
+    elif len(key) < _MIN_SECRET_KEY_LEN:
+        problems.append(f"SECRET_KEY 过短({len(key)} 字符),建议至少 {_MIN_SECRET_KEY_LEN} 位")
+    return problems
+
+
+def _check_database_url() -> list[str]:
+    problems: list[str] = []
+    url = settings.DATABASE_URL.lower()
+    for pat in _WEAK_DB_PATTERNS:
+        if pat in url:
+            problems.append(f"DATABASE_URL 使用示例弱密码({pat.rstrip('@')}),生产环境必须更换")
+            break
+    return problems
+
+
+async def validate_security() -> dict:
+    """执行安全校验。
+
+    返回 {"ok": bool, "errors": [...], "warnings": [...]}。
+    errors 为阻断性问题,warnings 为建议。
+    """
+    errors: list[str] = []
+    warnings: list[str] = []
+
+    errors.extend(_check_secret_key())
+    if not await _auth_configured():
+        errors.append("管理鉴权未配置:请设置 ADMIN_PASSWORD 或 ADMIN_API_KEY")
+    warnings.extend(_check_database_url())
+
+    # 生产环境:DB 弱密码也升级为阻断
+    if settings.APP_ENV == "production" and warnings:
+        errors.extend(warnings)
+        warnings = []
+
+    ok = not errors
+    return {"ok": ok, "errors": errors, "warnings": warnings}
+
+
+async def assert_security_on_startup() -> None:
+    """启动入口:生产环境校验失败则拒绝启动,开发环境仅警告。"""
+    result = await validate_security()
+
+    for w in result["warnings"]:
+        logger.warning("[security-check] %s", w)
+
+    if result["ok"]:
+        if result["warnings"]:
+            logger.warning("[security-check] 存在 %d 项警告,建议修复", len(result["warnings"]))
+        else:
+            logger.info("[security-check] 安全校验通过")
+        return
+
+    # 阻断
+    is_prod = settings.APP_ENV == "production"
+    level = logging.ERROR if is_prod else logging.WARNING
+    for e in result["errors"]:
+        logger.log(level, "[security-check] %s", e)
+
+    if is_prod:
+        logger.critical(
+            "[security-check] 生产环境安全校验失败,拒绝启动。请修复上述 %d 项问题后重试。",
+            len(result["errors"]),
+        )
+        # 明确退出,避免带着危险配置上线
+        sys.exit(1)
+
+    logger.warning("[security-check] 开发环境存在 %d 项问题(未阻断),请尽快修复", len(result["errors"]))