SFTPClient类
📅 2026/7/22 10:10:57
👁️ 阅读次数
📝 编程学习
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 输出信息 =====
[+] 任务完成
编程学习
技术分享
实战经验