qinglong
29 lines
968 B
Python
29 lines
968 B
Python
from aiohttp import web
|
|
from ..utils.auth import panel_auth
|
|
import re
|
|
|
|
# Shell 元字符黑名单 — 防止命令注入
|
|
_SHELL_DANGER = re.compile(r'[;&|`$(){}!#~<>]')
|
|
|
|
|
|
def setup_routes(app, prefix=''):
|
|
app.router.add_post(f'{prefix}/api/command', panel_auth(exec_cmd))
|
|
|
|
|
|
async def exec_cmd(req):
|
|
d = await req.json()
|
|
raw = d.get('command', '')
|
|
# 安全检查: 拒绝含 shell 元字符的命令
|
|
if _SHELL_DANGER.search(raw):
|
|
return web.json_response(
|
|
{"success": False, "error": "命令包含不允许的字符"}, status=400
|
|
)
|
|
cs = req.app.get('service_manager').get_service("command")
|
|
if not cs:
|
|
return web.json_response({"error": "Missing"}, status=503)
|
|
try:
|
|
res = await cs.process_command(raw, source="web")
|
|
return web.json_response({"success": True, "output": str(res)})
|
|
except Exception:
|
|
return web.json_response({"success": False, "error": "命令执行失败"})
|