SFTPClient类

SFTPClient类

importparamikoaspk host='10.1.8.128'port=22user='root'password='root'try:#建立隧道t=pk.Transport((host,port))#建立连接t.connect(username=user,password=password)#客户端sftp=pk.SFTPClient.from_transport(t)#上传文件sftp.put('C:\\Users\\孟宇\\PyCharmMiscProject\\paramiko模块\\system.log','/root/system.log')print('成功！')t.close()exceptExceptionase:print(e)成功！

sftp.get('/root/abc.txt','./abc.txt')print('成功')成功

sftp.mkdir('/root/test')print('成功')成功

sftp.rename('/root/test','/root/test_python')print('成功')成功

#删除目录sftp.rmdir('/root/test_python')#删除文件sftp.remove('/root/abc.txt')print('成功')t.close()

#查看文件状态print(sftp.stat('/root/system.log'))#查看文件列表print(sftp.listdir('/root/'))-rw-r--r--1003459720Jul09:26?['.bash_logout','.bash_profile','.bashrc','.cshrc','.tcshrc','anaconda-ks.cfg','.bash_history','.viminfo.tmp','Python-3.14.2.tgz','Python-3.14.2','.config','.cache','system.log','.viminfo']

堡垒机模式下的远程命令执行

importsysimportparamiko as pkimporttime# ===== 配置信息 =====blhost='10.1.8.128'# 堡垒机 IPbluser='root'blpwd='root'host='10.1.8.129'# 业务机 IPuser='root'pwd='root'port=22# ===== 开始连接 =====pk.util.log_to_file('sshclient_system.log')sshclient=pk.SSHClient()sshclient.set_missing_host_key_policy(pk.AutoAddPolicy())try: sshclient.connect(hostname=blhost,username=bluser,password=blpwd,timeout=10)print("[+] 堡垒机连接成功")except Exception as e: print(f"[-] 堡垒机连接失败: {e}")sys.exit(1)# 创建交互式 shellchannel=sshclient.invoke_shell()channel.settimeout(10)# 发送 ssh 登录业务机命令channel.send(f'ssh {user}@{host}\n')buff=''# ----- 阶段1：等待密码提示或 yes/no -----whileTrue: try: resp=channel.recv(9999).decode('utf-8')ifnot resp:continuebuff+=resp print(f"[DEBUG] 收到: {repr(resp)}")except Exception as e: print(f"[-] 接收超时/错误: {e}")channel.close()sshclient.close()sys.exit(1)if'yes/no'inbuff.lower(): channel.send('yes\n')buff=''continueif'password:'inbuff.lower(): print("[+] 检测到密码提示，发送密码")break# ----- 阶段2：发送密码 -----channel.send(pwd +'\n')buff=''# ----- 阶段3：等待业务机 shell 提示符 -----whileTrue: try: resp=channel.recv(9999).decode('utf-8')ifnot resp:continuebuff+=resp print(f"[DEBUG] 收到: {repr(resp)}")except Exception as e: print(f"[-] 接收超时/错误: {e}")channel.close()sshclient.close()sys.exit(1)if'permission denied'inbuff.lower(): print("[-] 业务机认证失败，请检查密码")channel.close()sshclient.close()sys.exit(1)if'#'inbuff or'$'inbuff: print("[+] 业务机登录成功")break# ----- 阶段4：执行 ifconfig 并读取输出 -----channel.send('ifconfig\n')buff=''time.sleep(0.5)whileTrue: try: resp=channel.recv(9999).decode('utf-8')ifnot resp:continuebuff+=respif'#'inbuff or'$'inbuff:breakexcept Exception as e: print(f"[-] 接收命令输出超时: {e}")break# 打印输出（去除最后一行提示符）lines=buff.splitlines()iflines and('#'inlines[-1]or'$'inlines[-1]): lines.pop()print("\n===== ifconfig 输出 =====\n")print('\n'.join(lines))channel.close()sshclient.close()ifconfigens160:flags=4163<UP,BROADCAST,RUNNING,MULTICAST>mtu1500inet10.1.8.129 netmask255.255.255.0 broadcast10.1.8.255 inet6 fe80::20c:29ff:feb5:6649 prefixlen64scopeid 0x20<link>ether 00:0c:29:b5:66:49 txqueuelen1000(Ethernet)RX packets350bytes164182(160.3KiB)RX errors0dropped0overruns0frame0TX packets372bytes40988(40.0KiB)TX errors0dropped0overruns0carrier0collisions0lo:flags=73<UP,LOOPBACK,RUNNING>mtu65536inet127.0.0.1 netmask255.0.0.0 inet6 ::1 prefixlen128scopeid 0x10<host>loop txqueuelen1000(Local Loopback)RX packets0bytes0(0.0B)RX errors0dropped0overruns0frame0TX packets0bytes0(0.0B)TX errors0dropped0overruns0carrier0collisions0

远程文件上传

importparamikoimportsysimporttime# ===== 堡垒机信息 =====blip='10.1.8.128'bluser='root'blpwd='root'# ===== 业务服务器信息 =====hostname='10.1.8.129'username='root'password='root'# ===== 路径配置 =====tmpdir='/tmp'remotedir='/data'localpath='C:\\Users\\孟宇\\Downloads\\Python-3.14.2.tgz'tmppath=tmpdir+'/Python-3.14.2.tgz'remotepath=remotedir+'/Python-3.14.2.tgz'port=22PASS_PROMPT='password:'SHELL_PROMPT='#'# 业务机提示符paramiko.util.log_to_file('syslogin.log')# ----- 阶段1：上传文件到堡垒机 -----t=paramiko.Transport((blip,port))t.connect(username=bluser,password=blpwd)sftp=paramiko.SFTPClient.from_transport(t)print("[+] 上传本地文件到堡垒机...")sftp.put(localpath,tmppath)print("[+] 上传完成")sftp.close()t.close()# ----- 阶段2：通过堡垒机执行 scp -----ssh=paramiko.SSHClient()ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())ssh.connect(hostname=blip,username=bluser,password=blpwd,timeout=10)print("[+] 堡垒机 SSH 连接成功")channel=ssh.invoke_shell()channel.settimeout(10)time.sleep(0.5)# ---------- 2.1 先创建业务机上的目标目录 ----------mkdir_cmd=f'ssh{username}@{hostname}"mkdir -p{remotedir}"\n'channel.send(mkdir_cmd)print(f"[+] 发送创建目录命令:{mkdir_cmd.strip()}")buff=''whileTrue:try:resp=channel.recv(9999).decode('utf-8')ifnotresp:continuebuff+=respprint(f"[DEBUG] mkdir 输出:{repr(resp)}")exceptExceptionase:print(f"[-] 接收超时:{e}")channel.close()ssh.close()sys.exit(1)# 如果出现密码提示（可能业务机要求密码），处理ifPASS_PROMPTinbuff.lower():channel.send(password+'\n')buff=''continue# 如果出现提示符，说明命令执行完成if'#'inbuffor'$'inbuff:print("[+] 目录创建完成（或已存在）")break# ---------- 2.2 发送 scp 命令 ----------scp_cmd=f'scp{tmppath}{username}@{hostname}:{remotepath}\n'channel.send(scp_cmd)print(f"[+] 发送 scp 命令:{scp_cmd.strip()}")buff=''# 等待密码提示（如果是首次 scp 会询问）whileTrue:try:resp=channel.recv(9999).decode('utf-8')ifnotresp:continuebuff+=respprint(f"[DEBUG] scp 输出:{repr(resp)}")exceptExceptionase:print(f"[-] 接收超时:{e}")channel.close()ssh.close()sys.exit(1)# 处理主机密钥确认if'yes/no'inbuff.lower():channel.send('yes\n')buff=''continue# 检测密码提示ifPASS_PROMPTinbuff.lower():print("[+] 检测到密码提示，发送业务机密码")channel.send(password+'\n')buff=''continue# 如果出现提示符，传输完成if'#'inbuffor'$'inbuff:print("[+] scp 传输完成")break# 打印 scp 输出（去除提示符行）lines=buff.splitlines()iflinesand('#'inlines[-1]or'$'inlines[-1]):lines.pop()print("\n===== scp 输出信息 =====\n")print('\n'.join(lines))channel.close()ssh.close()print("[+] 任务完成")[+]上传本地文件到堡垒机...[+]上传完成[+]堡垒机 SSH 连接成功[+]发送创建目录命令:ssh root@10.1.8.129"mkdir -p /data"[DEBUG]mkdir 输出:'Last login: Mon Jul 20 14:05:12 2026 from 10.1.8.1\r\r\n[\x1b[91mroot\x1b[93m@\x1b[92;1mxhde\x1b[0m \x1b[94m~\x1b[0m \x1b[35m14:06:02\x1b[0m]\x1b[93m#\x1b[0m '[+]目录创建完成（或已存在）[+]发送 scp 命令:scp/tmp/Python-3.14.2.tgz root@10.1.8.129:/data/Python-3.14.2.tgz[DEBUG]scp 输出:'ssh root@10.1.8.129 "mkdir -p /data"\r\n'[DEBUG]scp 输出:'scp /tmp/Python-3.14.2.tgz root@10.1.8.129:/data/Python-3.14.2.tgz\r\n'[DEBUG]scp 输出:"\rroot@10.1.8.129's password: "[+]检测到密码提示，发送业务机密码[DEBUG]scp 输出:'\r\n'[DEBUG]scp 输出:'[\x1b[91mroot\x1b[93m@\x1b[92;1mxhde\x1b[0m \x1b[94m~\x1b[0m \x1b[35m14:06:03\x1b[0m]\x1b[93m#\x1b[0m '[+]scp 传输完成=====scp 输出信息=====[+]任务完成

输出: ‘scp /tmp/Python-3.14.2.tgz root@10.1.8.129:/data/Python-3.14.2.tgz\r\n’
[DEBUG] scp 输出: "\rroot@10.1.8.129’s password: "
[+] 检测到密码提示，发送业务机密码
[DEBUG] scp 输出: ‘\r\n’
[DEBUG] scp 输出: '[\x1b[91mroot\x1b[93m@\x1b[92;1mxhde\x1b[0m \x1b[94m~\x1b[0m \x1b[35m14:06:03\x1b[0m]\x1b[93m#\x1b[0m ’
[+] scp 传输完成

===== scp 输出信息 =====

[+] 任务完成