122 lines
4.1 KiB
Python
122 lines
4.1 KiB
Python
"""生产环境启动安全校验:缺失关键配置则拒绝启动(生产)或警告(开发)。
|
|
|
|
校验项:
|
|
- 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"]))
|