Linux SSH 管理项目

pip install -r requirements.txt

from MySSHBase import MySSH

ssh = MySSH("8.140.234.178", "root", "密码", default\_port=22)

\# 方式一：手动管理

ssh.Init()

print(ssh.BatchCMD("uname -a"))

ssh.CloseSSH()

\# 方式二：with 上下文（自动连接、退出自动关闭）

with MySSH("8.140.234.178", "root", "密码", 22) as ssh:

    print(ssh.GetSystemVersion())

{

  "code": 200,

  "msg": "命令执行成功，已获取结果",

  "data": "Linux ...",

  "stderr": "",

  "exit\_status": 0,

  "timed\_out": false

}

\# sudo 需要密码 + 确认 的场景

res = ssh.InteractiveCMD("sudo systemctl restart nginx", [

    {"expect": r"\[sudo\].*password", "send": "我的密码"},

    {"expect": "Do you want to continue", "send": "y"},

])

print(res)   # 含完整输出、exit_status、每一步 matched 结果

session = ssh.InvokeCMD("sudo apt update")

session.expect(r"password.*:", timeout=10)   # 等待密码提示

session.send("我的密码")                      # 发密码并回车

exit_code = session.wait_exit()              # 等命令结束

print(session.get_output())                  # 拿完整输出

session.close()

\# 方式一：构造时开启（推荐）

ssh = MySSH("8.140.234.178", "root", "密码", 22, filter_mode="blacklist")

\# 方式二：运行时切换

ssh.SetCommandFilter("blacklist")

print(ssh.BatchCMD("rm -rf /tmp/x"))        # code=403，未执行

ssh.AddBlacklist("poweroff")                 # 追加自定义规则

ssh.SetCommandFilter("whitelist")            # 切换白名单

print(ssh.BatchCMD("ls /tmp"))               # 放行

print(ssh.BatchCMD("fdisk -l"))              # code=403，不在名单

from SSHSessionManager import SSHSessionManager

mgr = SSHSessionManager()

\# 新建窗体（不传 ssh 时内部新建连接；传已连接 MySSH 实例则复用连接）

r1 = mgr.CreateSession("8.140.234.178", "root", "密码", 22, name="窗口A", ssh=ssh)

r2 = mgr.CreateSession("8.140.234.178", "root", "密码", 22, name="窗口B")

mgr.ListSessions()                # 遍历：数量、激活窗体、每个窗体的信息与最新输出

mgr.SwitchSession("S2")           # 切入窗口B（窗口A切出后台保留）

mgr.Execute("S1", "echo hi")      # 指定窗体执行命令（默认执行在激活窗体）

mgr.Send("S2", "echo test")       # 向任意窗体发送输入

mgr.Read("S1", timeout=1)         # 读取任意窗体新输出

mgr.WaitPrompt("S2", pattern=r"#\s*$")   # 等待提示符，确认命令执行完毕

mgr.CloseSession("S2")            # 关闭单个窗体

mgr.CloseAll()                    # 全部关闭（新建的连接随之释放）

mgr.Execute("S3", "sleep 3; echo 后台任务已完成", wait=1)  # 启动长任务
mgr.SwitchSession("S4")                                   # 立即切出
# ... 在 S4 正常干活 ...
mgr.SwitchSession("S3")                                   # 切回
mgr.Read("S3")   # 输出已在缓冲区："... 后台任务已完成 [root@... ~]#"

mgr.GetCommandState("S3")
\# -> {"state": "idle"}  空闲（提示符就绪）
\# -> {"state": "busy"}  正在执行（探针未回显）
\# -> {"state": "closed"} 已关闭

mgr.WaitCommandFinish("S3", timeout=30)   # 阻塞等待当前命令执行完毕

mgr.PostMessage("S3", "你好，窗口C", from_id="窗口D")   # 投递到收件箱（不干扰终端输入）
mgr.Broadcast("系统通知：稍后维护")                      # 广播给所有会话
mgr.ReadMessages("S3")                                 # 读取（默认读后清空）收件箱
mgr.SendToSession("S4", "echo hi", from_session_id="S3")  # 直接注入输入到另一窗体并回车执行

mgr.ExportSessions("backup.json")    # 导出所有会话的连接配置 + 命令历史
mgr.CloseAll()                        # 模拟重启/退出
mgr.RestoreSessions("backup.json")   # 重新连接并恢复（名称与历史保留）
mgr.RestoreSessions("backup.json", connect=False)   # dry-run，只解析校验不连接
mgr.GetHistory("S4")                  # 查看任意会话的命令历史

from AdminDataBase import AdminDataBase

db = AdminDataBase("hosts.db")

db.InitDatabase()

db.AddHost(address="8.140.234.178", username="root", password="密码", port="22")

print(db.ShowHostList())

db.AddHostGroup("WebServers")

db.AddHostGroupOnUUID("WebServers", "\<uuid>")

print(db.PingAllHosts())   # 真实 TCP 连通检测

pip install -r requirements.txt
python mcp.py                      # 默认 streamable-http，监听 127.0.0.1:8001
python mcp.py --port 9001 --log-level DEBUG
python mcp.py --config config.json # 可选配置文件覆盖默认值
python mcp.py --list-tools         # 仅打印已注册工具清单后退出