fix(security): 代码审查安全加固 — 三批次修复(6H/13M/12L)
第一批(快速修复): - H6: main.py --debug 默认值 True→False,防止 Werkzeug RCE - M2+M3: /search 增加 validate_query + top_k 范围限制(1-50) - M4: context_count 范围限制(0-10) + 异常捕获 - L3: assert → raise RuntimeError(生产环境 API Key 检查) - H1: SSE 错误事件移除 traceback 字段 第二批(安全加固): - H2+H3: 文档接口路径遍历 realpath 校验 + 文件类型/大小限制 - H4+H5: 批量上传文件大小检查 - M6: LIKE 查询通配符转义 - M1: 37 处 str(e) 异常信息统一脱敏(6 文件) - M5: CORS 生产环境限制来源 - M7: SESSION_MANAGER None 保护(503) - M11: subprocess 参数注入防护(白名单 + -- 分隔符) 第三批(架构改进): - M8+M9: 提取 JSON 解析共享工具(extract_json_object/list) - M10: Prompt 注入检测防御(prompt_guard.py) - M12: 解析器文件大小限制(Excel 50MB/TXT 20MB/PDF 100MB) - M13: 全局单例竞态条件双重检查锁定(engine/bm25/intent_analyzer)
This commit is contained in:
@@ -237,6 +237,76 @@ def parse_json_list_from_response(content: str) -> Optional[List[dict]]:
|
||||
return None
|
||||
|
||||
|
||||
def extract_json_object(content: str) -> Optional[dict]:
|
||||
"""
|
||||
多策略从 LLM 响应中提取 JSON 对象(增强版)
|
||||
|
||||
在 parse_json_from_response 基础上增加 fallback 策略:
|
||||
1. 先调用 parse_json_from_response(markdown 代码块 → 直接解析)
|
||||
2. 失败后 fallback 到正则匹配最外层 {...} 块
|
||||
|
||||
Args:
|
||||
content: LLM 返回的原始内容
|
||||
|
||||
Returns:
|
||||
解析后的字典,全部策略失败返回 None
|
||||
"""
|
||||
if not content:
|
||||
return None
|
||||
|
||||
# 策略1+2:markdown 代码块提取 + 直接 json.loads
|
||||
result = parse_json_from_response(content)
|
||||
if result is not None and isinstance(result, dict):
|
||||
return result
|
||||
|
||||
# 策略3(fallback):正则匹配最外层 JSON 对象 {...}
|
||||
brace_match = re.search(r'\{[\s\S]*\}', content)
|
||||
if brace_match:
|
||||
try:
|
||||
parsed = json.loads(brace_match.group(0))
|
||||
if isinstance(parsed, dict):
|
||||
return parsed
|
||||
except (json.JSONDecodeError, TypeError, ValueError):
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def extract_json_list(content: str) -> Optional[list]:
|
||||
"""
|
||||
多策略从 LLM 响应中提取 JSON 数组(增强版)
|
||||
|
||||
在 parse_json_list_from_response 基础上增加 fallback 策略:
|
||||
1. 先调用 parse_json_list_from_response(markdown 代码块 → 直接解析 → 嵌套提取)
|
||||
2. 失败后 fallback 到正则匹配最外层 [...] 块
|
||||
|
||||
Args:
|
||||
content: LLM 返回的原始内容
|
||||
|
||||
Returns:
|
||||
解析后的列表,全部策略失败返回 None
|
||||
"""
|
||||
if not content:
|
||||
return None
|
||||
|
||||
# 策略1+2:markdown 代码块提取 + 直接 json.loads + 嵌套 key 提取
|
||||
result = parse_json_list_from_response(content)
|
||||
if result is not None:
|
||||
return result
|
||||
|
||||
# 策略3(fallback):正则匹配最外层 JSON 数组 [...]
|
||||
bracket_match = re.search(r'\[[\s\S]*\]', content)
|
||||
if bracket_match:
|
||||
try:
|
||||
parsed = json.loads(bracket_match.group(0))
|
||||
if isinstance(parsed, list):
|
||||
return parsed
|
||||
except (json.JSONDecodeError, TypeError, ValueError):
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
|
||||
# ==================== 便捷函数 ====================
|
||||
|
||||
def quick_ask(
|
||||
|
||||
Reference in New Issue
Block a user