diff --git a/mcp_center/servers/oe_cli_mcp_server/mcp_tools/base_tools/network_tools/base.py b/mcp_center/servers/oe_cli_mcp_server/mcp_tools/base_tools/network_tools/base.py index 4ac62a7393993ef047e198f8f46cbca33bb25e1a..ac29b6ca5fc948fb59d5053785a8d26a6345c6fb 100644 --- a/mcp_center/servers/oe_cli_mcp_server/mcp_tools/base_tools/network_tools/base.py +++ b/mcp_center/servers/oe_cli_mcp_server/mcp_tools/base_tools/network_tools/base.py @@ -1,4 +1,5 @@ import logging +import os import subprocess from typing import Dict, Optional, List @@ -6,6 +7,7 @@ import paramiko from config.public.base_config_loader import BaseConfig, LanguageEnum + logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) @@ -54,6 +56,16 @@ def _open_ssh(remote_auth: Dict) -> paramiko.SSHClient: return ssh +def _find_cmd_absolute_path(cmd: str) -> Optional[str]: + """查找命令的绝对路径(兼容特殊情况)""" + common_paths = ["/usr/bin", "/bin", "/usr/sbin", "/sbin", "/usr/local/bin"] + for path in common_paths: + cmd_path = os.path.join(path, cmd) + if os.path.exists(cmd_path) and os.access(cmd_path, os.X_OK): + return cmd_path + return None + + def _build_fix_command() -> str: """ 构造修复 ifcfg-ens3 中 BOOTPROTO 并重启 NetworkManager 的命令 @@ -62,6 +74,11 @@ def _build_fix_command() -> str: 2. 设置 BOOTPROTO=dhcp 3. systemctl restart NetworkManager """ + # 查找命令的绝对路径 + sed_path = _find_cmd_absolute_path("sed") or "sed" + grep_path = _find_cmd_absolute_path("grep") or "grep" + systemctl_path = _find_cmd_absolute_path("systemctl") or "systemctl" + ifcfg_path = "/etc/sysconfig/network-scripts/ifcfg-ens3" parts: List[str] = [] # 确认配置文件存在 @@ -74,14 +91,14 @@ def _build_fix_command() -> str: ) # 规范化 BOOTPROTO 行:若存在则直接替换为 dhcp parts.append( - "sed -i -e 's/^[#]*[[:space:]]*BOOTPROTO=.*/BOOTPROTO=dhcp/' \"$IFCFG\"" + f"{sed_path} -i -e 's/^[#]*[[:space:]]*BOOTPROTO=.*/BOOTPROTO=dhcp/' \"$IFCFG\"" ) # 若不存在 BOOTPROTO 行,则追加一行 parts.append( - "grep -q '^[[:space:]]*BOOTPROTO=' \"$IFCFG\" || echo 'BOOTPROTO=dhcp' >> \"$IFCFG\"" + f"{grep_path} -q '^[[:space:]]*BOOTPROTO=' \"$IFCFG\" || echo 'BOOTPROTO=dhcp' >> \"$IFCFG\"" ) # 重启 NetworkManager 服务 - parts.append("systemctl restart NetworkManager") + parts.append(f"{systemctl_path} restart NetworkManager") return " && ".join(parts) diff --git a/mcp_center/servers/oe_cli_mcp_server/mcp_tools/base_tools/ssh_fix_tool/base.py b/mcp_center/servers/oe_cli_mcp_server/mcp_tools/base_tools/ssh_fix_tool/base.py index 1d06d15e421576240b5e87256881229f30c6b6f7..e15959de2b6d03b4954153b8b3f7a758908c63cc 100644 --- a/mcp_center/servers/oe_cli_mcp_server/mcp_tools/base_tools/ssh_fix_tool/base.py +++ b/mcp_center/servers/oe_cli_mcp_server/mcp_tools/base_tools/ssh_fix_tool/base.py @@ -1,4 +1,5 @@ import logging +import os import socket import subprocess from typing import Dict, Optional, List @@ -62,7 +63,26 @@ def ssh_port_ping(target: str, port: int = 22, timeout: int = 5) -> Dict: return result +def _find_cmd_absolute_path(cmd: str) -> Optional[str]: + """查找命令的绝对路径(兼容特殊情况)""" + common_paths = ["/usr/bin", "/bin", "/usr/sbin", "/sbin", "/usr/local/bin"] + for path in common_paths: + cmd_path = os.path.join(path, cmd) + if os.path.exists(cmd_path) and os.access(cmd_path, os.X_OK): + return cmd_path + return None + + def _run_local(cmd: List[str]) -> subprocess.CompletedProcess: + """执行本地命令,自动查找命令的绝对路径""" + # 查找命令的绝对路径 + cmd_name = cmd[0] + cmd_path = _find_cmd_absolute_path(cmd_name) + if not cmd_path: + raise FileNotFoundError(f"命令不存在:{cmd_name}") + + # 替换为绝对路径 + cmd[0] = cmd_path return subprocess.run( cmd, check=True, @@ -122,7 +142,9 @@ def check_sshd_status(target: Optional[str]) -> Dict: ssh: Optional[paramiko.SSHClient] = None try: ssh = _open_ssh(remote_auth) - stdin, stdout, stderr = ssh.exec_command("systemctl status sshd") + # 查找 systemctl 的绝对路径 + systemctl_path = _find_cmd_absolute_path("systemctl") or "systemctl" + stdin, stdout, stderr = ssh.exec_command(f"{systemctl_path} status sshd") out = stdout.read().decode("utf-8", errors="replace").strip() err = stderr.read().decode("utf-8", errors="replace").strip() if err and "Active:" not in out: @@ -178,22 +200,27 @@ def fix_sshd_config_and_restart(target: Optional[str]) -> Dict: - 若不存在目标键则追加 - 最后重启 sshd """ + # 查找命令的绝对路径 + sed_path = _find_cmd_absolute_path("sed") or "sed" + grep_path = _find_cmd_absolute_path("grep") or "grep" + systemctl_path = _find_cmd_absolute_path("systemctl") or "systemctl" + sed_parts = [] # 替换或解注释三项配置 sed_parts.append( - r"sed -i -e 's/^[#]*[[:space:]]*Port[[:space:]].*/Port 22/' " - r"-e 's/^[#]*[[:space:]]*PermitRootLogin[[:space:]].*/PermitRootLogin yes/' " - r"-e 's/^[#]*[[:space:]]*PasswordAuthentication[[:space:]].*/PasswordAuthentication yes/' " - r"/etc/ssh/sshd_config" + f"{sed_path} -i -e 's/^[#]*[[:space:]]*Port[[:space:]].*/Port 22/' " + f"-e 's/^[#]*[[:space:]]*PermitRootLogin[[:space:]].*/PermitRootLogin yes/' " + f"-e 's/^[#]*[[:space:]]*PasswordAuthentication[[:space:]].*/PasswordAuthentication yes/' " + f"/etc/ssh/sshd_config" ) # 若 key 不存在则追加 for k, line in desired_lines.items(): sed_parts.append( - f"grep -q '^{k}[[:space:]]' /etc/ssh/sshd_config || " + f"{grep_path} -q '^{k}[[:space:]]' /etc/ssh/sshd_config || " f"echo '{line}' >> /etc/ssh/sshd_config" ) # 重启 sshd - sed_parts.append("systemctl restart sshd") + sed_parts.append(f"{systemctl_path} restart sshd") # 组合为单条 shell 命令 return " && ".join(sed_parts)