111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
"""推公钥到远程服务器 /root/.ssh/authorized_keys。
|
|
|
|
用法:
|
|
set REMOTE_PASS=xxx (Windows PowerShell: $env:REMOTE_PASS='xxx')
|
|
python scripts/push_ssh_key.py
|
|
|
|
依赖:paramiko
|
|
|
|
密钥优先级:id_ed25519.pub > id_rsa.pub(可由环境变量 SSH_PUB_KEY_PATH 覆盖)
|
|
|
|
去重策略:用 SSH 公钥 fingerprint(SHA256)匹配,而不是用 marker。
|
|
这样同一台机器多次推、不同算法/不同机器都能正确去重。
|
|
"""
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
import paramiko
|
|
|
|
HOST = "207.57.129.228"
|
|
PORT = 19717
|
|
USER = "root"
|
|
PASSWORD = os.environ["REMOTE_PASS"] # 由调用方设置
|
|
# 优先 ed25519,fallback 到 rsa;亦可通过 SSH_PUB_KEY_PATH 指定
|
|
_PUB_KEY_CANDIDATES = ("~/.ssh/id_ed25519.pub", "~/.ssh/id_rsa.pub")
|
|
_candidate = os.environ.get("SSH_PUB_KEY_PATH")
|
|
if _candidate:
|
|
PUB_KEY_PATH = os.path.expanduser(_candidate)
|
|
else:
|
|
PUB_KEY_PATH = None
|
|
for _p in _PUB_KEY_CANDIDATES:
|
|
_expanded = os.path.expanduser(_p)
|
|
if os.path.exists(_expanded):
|
|
PUB_KEY_PATH = _expanded
|
|
break
|
|
if PUB_KEY_PATH is None:
|
|
PUB_KEY_PATH = os.path.expanduser(_PUB_KEY_CANDIDATES[0]) # 默认仍给一个清晰的报错路径
|
|
|
|
|
|
def main() -> int:
|
|
pub = open(PUB_KEY_PATH, encoding="utf-8").read().strip()
|
|
print(f"公钥({PUB_KEY_PATH})前 60 字符: {pub[:60]}...")
|
|
|
|
# 本地算 fingerprint(SSH 公钥的 SHA256,OpenSSH 标准格式)
|
|
fp_proc = subprocess.run(
|
|
["ssh-keygen", "-lf", PUB_KEY_PATH],
|
|
capture_output=True, text=True, check=True,
|
|
)
|
|
fingerprint = fp_proc.stdout.split()[1] # 形如 "SHA256:xxxx"
|
|
print(f"fingerprint: {fingerprint}")
|
|
|
|
client = paramiko.SSHClient()
|
|
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
try:
|
|
client.connect(
|
|
HOST, port=PORT, username=USER, password=PASSWORD,
|
|
timeout=15, allow_agent=False, look_for_keys=False,
|
|
)
|
|
except Exception as e:
|
|
print(f"连接失败: {e}", file=sys.stderr)
|
|
return 1
|
|
|
|
# 1) 检查 .ssh 目录
|
|
cmds = [
|
|
"mkdir -p /root/.ssh && chmod 700 /root/.ssh",
|
|
"touch /root/.ssh/authorized_keys && chmod 600 /root/.ssh/authorized_keys",
|
|
]
|
|
for c in cmds:
|
|
_exec(client, c)
|
|
|
|
# 2) 用 fingerprint 去重 — 服务器上是否已有这个 key
|
|
_si, _so, _se = client.exec_command(
|
|
f"grep -F '{fingerprint}' /root/.ssh/authorized_keys 2>/dev/null || true"
|
|
)
|
|
existing = _so.read().decode().strip()
|
|
if existing:
|
|
print(f"该 fingerprint 已在 authorized_keys,跳过: {existing.splitlines()[0][:80]}...")
|
|
else:
|
|
# 3) 追加(用 printf %s\\n 避免 echo 的换行问题)
|
|
marker = "news-deploy-key"
|
|
if marker not in pub:
|
|
pub_with_marker = f"{pub} {marker}"
|
|
else:
|
|
pub_with_marker = pub
|
|
quoted = pub_with_marker.replace("'", "'\\''")
|
|
_exec(client, f"printf '%s\\n' '{quoted}' >> /root/.ssh/authorized_keys")
|
|
print(f"已追加公钥到 /root/.ssh/authorized_keys({fingerprint})")
|
|
|
|
# 4) 验证权限
|
|
_exec(client, "ls -la /root/.ssh/ /root/.ssh/authorized_keys")
|
|
# 5) 列一下当前所有 key
|
|
_exec(client, "awk '{print NR, $1, $2, $3}' /root/.ssh/authorized_keys")
|
|
|
|
client.close()
|
|
return 0
|
|
|
|
|
|
def _exec(client: paramiko.SSHClient, cmd: str) -> None:
|
|
print(f"$ {cmd}")
|
|
_si, so, se = client.exec_command(cmd, timeout=15)
|
|
out = so.read().decode().strip()
|
|
err = se.read().decode().strip()
|
|
if out:
|
|
print(out)
|
|
if err:
|
|
print(f"[stderr] {err}", file=sys.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|