mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 16:00:43 +00:00
xmz7891
5fc2f97be7
用户买了 AWS Lightsail（2G 内存 / 2 vCPU / 60G SSD / 公网 IP
13.54.30.196），中转从随身 WiFi 那台小机器搬过去。
== 为什么这次搬迁解决了一堆老问题 ==
原机器（Debian aarch64、460MB、走 4G）一直有几件搞不定的：
* 线路对 Cloudflare 隧道的 **7844 端口间歇性阻断**（443 通、7844 不通），
隧道每 3~10 分钟掉一次、公网跟着 530 —— 查了很久才从
cloudflared 自己的 precheck 里看清
* 只有 460MB 内存
* 会掉电硬挂（实测 2026-10-08 18:42 一次非正常关机）
新机器**有公网 IP**，中转直接监听，不经隧道、不过 7844。
实测公网延迟从 **1000~2000 ms 降到 1.1 ms**。
== 代码改动 ==
1. **relay.py 的 DATA_DIR 默认值不再写死**
原来是 /root/lanremote-relay。新机器上服务以 admin 普通用户跑
（它没理由要 root），而 /root 是 0700 —— 部署时直接
PermissionError: '/root/lanremote-relay'。
改成"本脚本所在目录下的 data/"，搬到哪都跟着走。
2. **客户端支持"没有证书的中转"**
新服务器是明文 HTTP（没签证书），而两个客户端原来都往 https 走：
* 手表端：裸 IP 会被强制补 https:// -> 永远连不上，
报的还是 SSL 错误，很容易让人白查半天证书
* Android 9+ 默认禁止明文流量，没 usesCleartextTraffic
请求会被系统直接掐掉
Windows 端本来就是对的（正则里把任意纯 IP 都当"用明文"）。
现在手表端规则和它一致，并加了 usesCleartextTraffic。
3. 清掉电脑端一段函数内的重复代码（return 之后还有一份拷贝，
缩进仍在函数体内所以不报错，但读起来很迷惑）。
== 新服务器上的部署 ==
/opt/ykxmz-relay/ 服务端（relay/admin/landing/showcase）
ykxmz-relay.service 监听 0.0.0.0:13245
ykxmz-monitor.service/.timer 每 60 秒查公网，连续 3 次不通就叫预案
rescue-lightsail.sh 四级阶梯抢救（重启中转 / 强杀重拉 /
查防火墙 / 重启整机[默认关]）
预案针对新架构重写过：原来围着隧道转（重启隧道、重施 MSS/MTU），
现在没有隧道，改成重启中转 + 查端口/防火墙。
全量回归 46 个阶段全部通过。
129 lines
4.0 KiB
Python
129 lines
4.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
r'''连 AWS Lightsail 那台新服务器的 SSH 助手。
|
||
|
||
== 为什么单独一个模块，不改 devssh.py ==
|
||
|
||
devssh.py 是连**随身 WiFi 那台 4G 小机器**的（192.168.2.156，密码登录）。
|
||
那台还在跑（网站暂时还挂在上面），不能动它的连接逻辑。
|
||
|
||
新服务器用**密钥**登录、IP 是公网 13.54.30.196，认证方式完全不同，
|
||
所以单独一个模块更清楚。
|
||
|
||
== 用法 ==
|
||
|
||
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
|
||
|
||
HOST = os.environ.get("YKXMZ_LS_HOST", "13.54.30.196")
|
||
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 "不通"))
|