mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 12:30:44 +00:00
xmz7891
0ad3f648c6
上一个提交把仓库里的服务器 IP 换成了占位符，但**那些脚本本身要用这个 IP
去连服务器** —— 换掉之后它们就跑不起来了。这个提交把取值方式理清楚。
== 取值统一走 _build/lssh.py ==
YKXMZ_LS_HOST 环境变量
或 _build/.lightsail_host（一行 IP，已 gitignore）
都没有就报错，并告诉你怎么设
十个运维脚本改成 `from lssh import ... HOST as LS_HOST`，
公开仓库里一个具体地址都不留。
== 顺带修的两个自己造的 bug ==
* 用正则往 `from lssh import ...` 后面追加东西时，
追加到了行尾的 `# noqa: E402` **注释里面** ——
十处导入全是废的，脚本一跑就 NameError。
改成先剥掉注释、拼好导入、再把 noqa 加回去。
* 同一个正则还把 `from lssh import HOST as LS_HOST` 加进了
**lssh.py 自己**，造成自我导入。
教训：**正则批量改导入语句很危险**（注释、行尾、顺序都能坑到），
这种地方宁可逐个 edit。
== 为什么要重建产物 ==
docs/AFTER-INSTALL.txt 会被打进服务端 zip，而它上一轮改过
（去掉了写死的旧管理员密码）—— 不重打的话发出去的还是旧的。
重打后确认：三个产物里都没有旧密码、没有服务器 IP。
== 验证 ==
全量回归退出码 0，41 个阶段全 PASS
603 个 Python 文件语法通过
ykxmz-agent / server / watch 三个产物扫描干净
提交前泄漏闸门通过
153 lines
4.8 KiB
Python
153 lines
4.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
r'''连 AWS Lightsail 那台新服务器的 SSH 助手。
|
||
|
||
== 为什么单独一个模块，不改 devssh.py ==
|
||
|
||
devssh.py 是连**随身 WiFi 那台 4G 小机器**的（192.168.2.156，密码登录）。
|
||
那台还在跑（网站暂时还挂在上面），不能动它的连接逻辑。
|
||
|
||
新服务器用**密钥**登录、IP 是公网 <源站IP>，认证方式完全不同，
|
||
所以单独一个模块更清楚。
|
||
|
||
== 用法 ==
|
||
|
||
from lssh import run, upload, download, probe
|
||
run("uname -a") # 跑命令
|
||
upload("relay/relay.py", "/opt/ykxmz-relay/relay.py")
|
||
'''
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import socket
|
||
from pathlib import Path
|
||
|
||
import paramiko
|
||
|
||
|
||
def _host() -> str:
|
||
"""服务器公网 IP —— **不写死在代码里**。
|
||
|
||
这个仓库是公开的。IP 写进来等于把源站地址告诉所有人，
|
||
而源站躲在 Cloudflare 后面本来就不该被知道（直接打 IP 就绕过 CDN
|
||
和防护了）。2026-10-08 就因为 IP 泄漏返工过一次。
|
||
|
||
取值顺序：
|
||
1. 环境变量 YKXMZ_LS_HOST
|
||
2. _build/.lightsail_host（一行 IP，已 gitignore）
|
||
"""
|
||
env = os.environ.get("YKXMZ_LS_HOST", "").strip()
|
||
if env:
|
||
return env
|
||
f = Path(__file__).resolve().parent / ".lightsail_host"
|
||
if f.is_file():
|
||
return f.read_text(encoding="utf-8").strip()
|
||
raise SystemExit(
|
||
"没拿到服务器地址。二选一：\n"
|
||
" set YKXMZ_LS_HOST=<你的服务器IP>\n"
|
||
" 或把 IP 写进 _build/.lightsail_host（该文件已在 .gitignore 里）")
|
||
|
||
|
||
HOST = _host()
|
||
USER = os.environ.get("YKXMZ_LS_USER", "admin")
|
||
PORT = int(os.environ.get("YKXMZ_LS_PORT", "22"))
|
||
KEY = Path(os.environ.get(
|
||
"YKXMZ_LS_KEY", str(Path.home() / ".ssh" / "lightsail-ykxmz.pem")))
|
||
|
||
_client = None
|
||
|
||
|
||
def _connect(timeout: int = 25):
|
||
global _client
|
||
if _client is not None:
|
||
try:
|
||
# 探一下还活着没
|
||
_client.exec_command("true", timeout=5)
|
||
return _client
|
||
except Exception: # noqa: BLE001
|
||
_client = None
|
||
|
||
if not KEY.is_file():
|
||
raise RuntimeError(
|
||
"找不到 SSH 密钥：%s\n"
|
||
"（把 Lightsail 的 .pem 放到那里，或者设 YKXMZ_LS_KEY）" % KEY)
|
||
|
||
cli = paramiko.SSHClient()
|
||
cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||
cli.connect(HOST, port=PORT, username=USER,
|
||
key_filename=str(KEY), timeout=timeout,
|
||
allow_agent=False, look_for_keys=False,
|
||
banner_timeout=30, auth_timeout=30)
|
||
_client = cli
|
||
return cli
|
||
|
||
|
||
def run(command: str, timeout: int = 300, quiet: bool = False) -> tuple:
|
||
"""跑一条命令，返回 (退出码, 输出)。"""
|
||
cli = _connect()
|
||
if not quiet:
|
||
print(" [ls] %s" % command.splitlines()[0][:90])
|
||
stdin, stdout, stderr = cli.exec_command(command, timeout=timeout,
|
||
get_pty=False)
|
||
out = stdout.read().decode("utf-8", "replace")
|
||
err = stderr.read().decode("utf-8", "replace")
|
||
code = stdout.channel.recv_exit_status()
|
||
return code, (out + err)
|
||
|
||
|
||
def upload(local: str, remote: str, mode: int = 0o644) -> None:
|
||
"""传一个文件上去。"""
|
||
cli = _connect()
|
||
sftp = cli.open_sftp()
|
||
try:
|
||
# 先把远端目录建出来
|
||
d = os.path.dirname(remote)
|
||
if d:
|
||
try:
|
||
sftp.stat(d)
|
||
except IOError:
|
||
# 逐级建
|
||
parts = d.strip("/").split("/")
|
||
cur = ""
|
||
for p in parts:
|
||
cur += "/" + p
|
||
try:
|
||
sftp.stat(cur)
|
||
except IOError:
|
||
try:
|
||
sftp.mkdir(cur)
|
||
except IOError:
|
||
pass
|
||
sftp.put(local, remote)
|
||
sftp.chmod(remote, mode)
|
||
finally:
|
||
sftp.close()
|
||
print(" 已上传 %s -> %s" % (os.path.basename(local), remote))
|
||
|
||
|
||
def download(remote: str, local: str) -> None:
|
||
cli = _connect()
|
||
sftp = cli.open_sftp()
|
||
try:
|
||
Path(local).parent.mkdir(parents=True, exist_ok=True)
|
||
sftp.get(remote, local)
|
||
finally:
|
||
sftp.close()
|
||
print(" 已下载 %s -> %s" % (remote, local))
|
||
|
||
|
||
def probe() -> str | None:
|
||
"""通不通。返回一句话描述，或 None。"""
|
||
try:
|
||
code, out = run("uname -sm; nproc; free -m | sed -n 2p",
|
||
timeout=30, quiet=True)
|
||
if code != 0:
|
||
return None
|
||
return out.strip().replace("\n", " | ")
|
||
except Exception as exc: # noqa: BLE001
|
||
return "连接失败：%s" % exc
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print(" 目标: %s@%s:%d 密钥 %s" % (USER, HOST, PORT, KEY))
|
||
print(" " + (probe() or "不通"))
|