ssh - PythonIDE Docs

SSH 远程命令与 SFTP 文件传输：连接 Linux/树莓派等设备，执行命令、上传下载文件。

边界：需要可达的网络与有效凭据（密码或私钥）。不要把密码写进公开脚本；生产环境用 keychain 存凭据。连接和执行放在按钮回调；session_id 用完后务必 disconnect()。

#模块概览

项	说明
导入	`import ssh`
适合做什么	远程运维、部署脚本、拉取日志、SFTP 传文件
调用时机	用户点击连接/执行；长命令注意 `timeout`
推荐顺序	`connect()` → `execute()` / `upload()` / `download()` → `disconnect()`
会话	`connect()` 返回 `session_id`，后续 API 都要带上

#快速开始

下面脚本连接主机、执行命令并断开：

python

import ssh

HOST = "192.168.1.10"   # 换成你可访问的主机
USER = "pi"
PASSWORD = "YOUR_PASSWORD"  # 勿提交到版本库

sid = None
try:
    sid = ssh.connect(HOST, USER, password=PASSWORD)
    result = ssh.execute(sid, "echo hello")
    print(result["stdout"].strip())
    print("exit_code:", result["exit_code"])
except ssh.SSHError as exc:
    print("SSH 失败:", exc, f"code={exc.code}")
finally:
    if sid:
        ssh.disconnect(sid)

使用私钥认证：

python

import ssh

sid = ssh.connect(
    "example.com",
    "deploy",
    private_key=open("id_rsa").read(),
    passphrase="optional",
)

#AppUI 示例

连接与命令执行放在按钮回调；凭据存在 State（勿记录到日志）。

python

import appui
import ssh

state = appui.State(
    host="192.168.1.10",
    user="pi",
    password="",
    session_id="",
    output="—",
    status="填写主机信息后连接",
)


def do_connect():
    if not state.host or not state.user:
        state.status = "请填写主机和用户名"
        return
    if state.session_id:
        state.status = "已连接，请先断开"
        return
    try:
        sid = ssh.connect(
            state.host,
            state.user,
            password=state.password or None,
        )
        state.batch_update(
            session_id=sid or "",
            status="已连接",
            output="—",
        )
    except ssh.SSHError as exc:
        state.status = f"连接失败: {exc}"


def run_command():
    if not state.session_id:
        state.status = "请先连接"
        return
    try:
        result = ssh.execute(state.session_id, "uname -a", timeout=15)
        text = (result.get("stdout") or "").strip() or "（无输出）"
        code = result.get("exit_code", -1)
        state.batch_update(
            output=text[:500],
            status=f"exit_code={code}",
        )
    except ssh.SSHError as exc:
        state.status = f"执行失败: {exc}"


def do_disconnect():
    if not state.session_id:
        return
    try:
        ssh.disconnect(state.session_id)
        state.batch_update(
            session_id="",
            output="—",
            status="已断开",
        )
    except ssh.SSHError as exc:
        state.status = f"断开失败: {exc}"


def body():
    connected = bool(state.session_id)
    return appui.NavigationStack(
        appui.Form([
            appui.Section("连接", [
                appui.TextField("主机", text=state.host),
                appui.TextField("用户名", text=state.user),
                appui.SecureField("密码", text=state.password),
                appui.LabeledContent(
                    "会话",
                    value=state.session_id[:8] + "…" if connected else "未连接",
                ),
            ]),
            appui.Section("命令", [
                appui.Button(
                    "连接" if not connected else "已连接",
                    action=do_connect,
                ).button_style("bordered_prominent")
                .disabled(connected),
                appui.Button("执行 uname -a", action=run_command),
                appui.Button("断开", action=do_disconnect, role="destructive"),
            ]),
            appui.Section("输出", [
                appui.Text(state.output).font("caption"),
                appui.Text(state.status).foreground_color("secondaryLabel"),
            ], footer="需同一局域网或可达主机；密码勿写入公开代码。"),
        ]).navigation_title("SSH")
    )


appui.run(body, state=state)

#API 参考

#速查

API	作用
`connect(host, username, ...)`	建立连接 → `session_id`
`execute(session_id, command, timeout)`	远程执行 → `{stdout, exit_code}`
`upload(session_id, local, remote)`	SFTP 上传
`download(session_id, remote, local)`	SFTP 下载
`disconnect(session_id)`	关闭会话
`SSHError`	操作失败时抛出

#connect

connect(host, username, password=None, port=22, private_key=None, passphrase=None)

python

sid = ssh.connect("10.0.0.5", "admin", password="secret")
sid = ssh.connect("server", "deploy", port=2222, private_key=key_text)

密码与私钥至少提供一种。

#execute

execute(session_id, command, timeout=30)

python

result = ssh.execute(sid, "ls -la /tmp")
print(result["stdout"])
print(result["exit_code"])

stdout 为合并后的标准输出字符串；exit_code 为远程命令退出码。

#文件传输

python

ssh.upload(sid, "/local/file.txt", "/remote/dir/file.txt")
ssh.download(sid, "/remote/log.txt", "/local/log.txt")

upload 会上传到 remote_path 的父目录（Bridge 行为）。

#disconnect

disconnect(session_id) — 关闭并移除会话；后续再用同一 ID 会报 invalid_session。

#异常

`code`	含义
`invalid_input`	缺少 host/用户名/凭据
`invalid_session`	`session_id` 无效或已断开
`unknown_command`	Bridge 命令错误

#常见错误

错误写法	后果	修正
密码硬编码并提交 Git	凭据泄露	用 keychain 或运行时输入
不 `disconnect()`	连接泄漏	`finally` 里断开
在 `body()` 里 `connect()`	刷新时重复连接	放进按钮回调
主机不可达	连接超时	检查网络/VPN/防火墙

#相关文档

文档	用途
keychain	安全存储 SSH 凭据
network	HTTP 连通性检查
http_server	本地文件服务
原生能力入口	MiniApp 场景配方

#预期效果

运行示例后，界面应出现文档描述的目标结果；若与预期不符，先看「失败路径」并按返回值或日志排查。