qing
5dc5ec9af2
凭据不再进命令行 argv/会话日志，对齐 SKILLS_FORMAT.md J12 条例： - gitclient: token 改经 GIT_CONFIG_* 环境变量传 http.extraheader（git>=2.31），不再嵌入 clone URL（不进 .git/config）；新增 _scrub_token 输出统一脱敏 - ssh-exec: 移除 -p/--password，密码仅经环境变量 AIBOX_SSH_PASS 提供 - sftp-transfer: 移除 -p/--password，AIBOX_SFTP_PASS 优先、getpass 兜底，密钥认证不再强制索密码 - aes-crypto: 移除 --key，密钥仅经 --key-file/--load-key 文件途径；生成密钥不再打印明文；save_key_iv 写后 chmod 600 验证：py_compile 通过；aes 生成/加载/加解密闭环实测正常，--key 被拒绝；全量测试 493 passed / 4 failed（平台环境项）/ 1 skipped，与基线一致。审计报告 §17 结论 + §18 波次1修复记录同步。
87 lines
2.9 KiB
Python
Executable File
87 lines
2.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""ssh-exec - 通过SSH远程执行单条命令的工具（支持密码认证）"""
|
||
|
||
import argparse
|
||
import os
|
||
import sys
|
||
import paramiko
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description="ssh-exec - 通过SSH远程执行单条命令（支持密码认证）",
|
||
epilog="""
|
||
参数说明:
|
||
-H, --host SSH主机地址（默认: 127.0.0.1）
|
||
-P, --port SSH端口号（默认: 22）
|
||
-u, --user SSH用户名（必填）
|
||
-c, --command 要执行的远程命令（必填）
|
||
|
||
密码认证:
|
||
密码经环境变量 AIBOX_SSH_PASS 提供（必填），不走命令行参数，
|
||
避免凭据进入进程列表与会话日志（J12 条例）。
|
||
|
||
使用示例:
|
||
AIBOX_SSH_PASS=12345678 %(prog)s -u qing -c "ls -la"
|
||
AIBOX_SSH_PASS=mypass %(prog)s -u root -H 192.168.1.100 -P 22 -c "df -h"
|
||
AIBOX_SSH_PASS=12345678 %(prog)s -u qing -H 127.0.0.1 -c "uname -a"
|
||
""",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
)
|
||
parser.add_argument("-H", "--host", default="127.0.0.1", help="SSH主机地址（默认: 127.0.0.1）")
|
||
parser.add_argument("-P", "--port", type=int, default=22, help="SSH端口号（默认: 22）")
|
||
parser.add_argument("-u", "--user", required=True, help="SSH用户名（必填）")
|
||
parser.add_argument("-c", "--command", required=True, help="要执行的远程命令（必填）")
|
||
|
||
args = parser.parse_args()
|
||
|
||
# 密码仅从环境变量读取（S4：凭据不进命令行）
|
||
password = os.environ.get("AIBOX_SSH_PASS")
|
||
if not password:
|
||
print("错误: 请通过环境变量 AIBOX_SSH_PASS 提供SSH密码", file=sys.stderr)
|
||
sys.exit(2)
|
||
|
||
client = paramiko.SSHClient()
|
||
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||
|
||
try:
|
||
client.connect(
|
||
hostname=args.host,
|
||
port=args.port,
|
||
username=args.user,
|
||
password=password,
|
||
timeout=10,
|
||
)
|
||
|
||
stdin, stdout, stderr = client.exec_command(args.command, timeout=30)
|
||
|
||
exit_code = stdout.channel.recv_exit_status()
|
||
out_text = stdout.read().decode("utf-8", errors="replace")
|
||
err_text = stderr.read().decode("utf-8", errors="replace")
|
||
|
||
if out_text:
|
||
print(out_text, end="")
|
||
if err_text:
|
||
print(err_text, end="", file=sys.stderr)
|
||
|
||
sys.exit(exit_code)
|
||
|
||
except paramiko.AuthenticationException:
|
||
print(f"错误: SSH认证失败（{args.user}@{args.host}:{args.port}）", file=sys.stderr)
|
||
sys.exit(1)
|
||
except paramiko.SSHException as e:
|
||
print(f"错误: SSH连接失败 - {e}", file=sys.stderr)
|
||
sys.exit(1)
|
||
except TimeoutError:
|
||
print(f"错误: 连接超时（{args.host}:{args.port}）", file=sys.stderr)
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
print(f"错误: {e}", file=sys.stderr)
|
||
sys.exit(1)
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|