重构白名单路由机制,增加app前缀
This commit is contained in:
parent
fa42e6bb00
commit
01e6c63bc8
@ -57,19 +57,16 @@ async def authenticate_request(request: Request, allow_guest: bool) -> bool:
|
|||||||
logger.warning("认证失败: 未提供有效的认证信息")
|
logger.warning("认证失败: 未提供有效的认证信息")
|
||||||
return False
|
return False
|
||||||
|
|
||||||
@router.post("/{module_path:path}")
|
async def _process_whitelist_call(request: Request, full_module_path: str):
|
||||||
async def handle_request(request: Request, module_path: str):
|
"""通用处理:接收完整点分路径 '<package.module.function>' 并执行调用"""
|
||||||
"""
|
|
||||||
处理白名单路由请求
|
|
||||||
"""
|
|
||||||
try:
|
try:
|
||||||
# 解析模块路径并先导入模块,确保装饰器执行
|
# 解析模块路径并先导入模块,确保装饰器执行
|
||||||
module_info = parse_module_path(module_path)
|
module_info = parse_module_path(full_module_path)
|
||||||
module = import_module(module_info['module_path'])
|
module = import_module(module_info['module_path'])
|
||||||
|
|
||||||
# 模块导入后再检查是否是白名单函数
|
# 模块导入后再检查是否是白名单函数
|
||||||
if is_whitelisted(module_path):
|
if is_whitelisted(full_module_path):
|
||||||
whitelist_info = get_whitelisted_function(module_path)
|
whitelist_info = get_whitelisted_function(full_module_path)
|
||||||
func = whitelist_info['function']
|
func = whitelist_info['function']
|
||||||
|
|
||||||
# 检查HTTP方法
|
# 检查HTTP方法
|
||||||
@ -114,6 +111,23 @@ async def handle_request(request: Request, module_path: str):
|
|||||||
logger.error(f"Request handler error: {e}")
|
logger.error(f"Request handler error: {e}")
|
||||||
raise HTTPException(status_code=500, detail=str(e))
|
raise HTTPException(status_code=500, detail=str(e))
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/{module_path:path}")
|
||||||
|
async def handle_request(request: Request, module_path: str):
|
||||||
|
"""
|
||||||
|
兼容旧路径:直接传入完整点分路径 '<package.module.function>'
|
||||||
|
"""
|
||||||
|
return await _process_whitelist_call(request, module_path)
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/{app}/{module_path:path}")
|
||||||
|
async def handle_request_with_app_prefix(request: Request, app: str, module_path: str):
|
||||||
|
"""
|
||||||
|
新路径:支持以 app 作为前缀,例如 app.module.function
|
||||||
|
"""
|
||||||
|
full_module_path = f"{app}.{module_path}"
|
||||||
|
return await _process_whitelist_call(request, full_module_path)
|
||||||
|
|
||||||
def parse_module_path(module_path: str) -> Dict[str, str]:
|
def parse_module_path(module_path: str) -> Dict[str, str]:
|
||||||
"""解析模块路径"""
|
"""解析模块路径"""
|
||||||
parts = module_path.split('.')
|
parts = module_path.split('.')
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user