如何用纯Python实现无需Session Manager Plugin的AWS SSM SSH数据传输与端口转发

核心原理 #

SSM Session Manager本质是通过WebSocket连接与AWS SSM服务通信，再由SSM Agent在目标实例上转发流量。纯Python实现的关键是跳过官方Session Manager Plugin，直接调用SSM API创建会话、建立WebSocket通道，再在该通道上封装SSH协议并实现端口转发。

具体实现步骤 #

1. 调用SSM API创建会话

使用AWS SDK for Python（boto3）调用start_session接口，获取会话的WebSocket连接地址和身份令牌。需要指定目标实例ID和用于SSH的SSM文档（AWS-StartSSHSession）。

import boto3

# 初始化SSM客户端（需配置AWS凭证，可通过环境变量、~/.aws/credentials等方式）
ssm_client = boto3.client('ssm', region_name='你的AWS区域')

# 创建SSM会话
session_response = ssm_client.start_session(
    Target='i-xxxxxxxxx',  # 目标EC2实例ID
    DocumentName='AWS-StartSSHSession'
)

# 提取WebSocket连接所需参数
stream_url = session_response['StreamUrl']
session_token = session_response['TokenValue']

2. 建立WebSocket连接

用websocket-client库创建WebSocket连接，请求头中携带会话令牌进行身份验证。这个连接将作为SSH流量的传输通道。

import websocket

class SSMSocket:
    """封装WebSocket连接为socket兼容接口，供paramiko调用"""
    def __init__(self, stream_url, token):
        self.ws = websocket.create_connection(
            stream_url,
            header={"X-Session-Token": token}
        )
    
    def send(self, data):
        self.ws.send_binary(data)
        return len(data)
    
    def recv(self, bufsize):
        frame = self.ws.recv_frame()
        return frame.data if frame else b''
    
    def close(self):
        self.ws.close()

# 初始化SSM Socket
ssm_socket = SSMSocket(stream_url, session_token)

3. 在WebSocket通道上建立SSH连接

使用paramiko库（Python SSH实现），将自定义的SSM Socket传入Transport类，完成SSH身份验证（支持密码或密钥）。

import paramiko

# 初始化SSH Transport
transport = paramiko.Transport(ssm_socket)
transport.start_client()

# 身份验证（二选一）
# 方式1：密码验证
transport.auth_password(username='ec2-user', password='你的实例密码')
# 方式2：密钥验证
# private_key = paramiko.RSAKey.from_private_key_file('/path/to/your/key.pem')
# transport.auth_publickey(username='ec2-user', key=private_key)

4. 实现本地端口转发

通过监听本地端口，将本地连接的流量通过SSH会话转发到目标实例的指定端口。使用多线程处理双向数据流，确保转发的实时性。

import threading
from socket import socket, AF_INET, SOCK_STREAM

def forward_local_port(local_port, remote_port, transport):
    """将本地端口转发到目标实例的指定端口"""
    def handle_local_connection(local_sock):
        # 在SSH会话上创建TCP通道
        remote_sock = transport.open_channel(
            'direct-tcpip',
            ('localhost', remote_port),  # 目标实例的地址和端口
            local_sock.getpeername()
        )
        if not remote_sock:
            local_sock.close()
            return
        
        # 双向转发数据
        def forward(src, dst):
            while True:
                data = src.recv(1024)
                if not data:
                    break
                dst.send(data)
            src.close()
            dst.close()
        
        threading.Thread(target=forward, args=(local_sock, remote_sock)).start()
        threading.Thread(target=forward, args=(remote_sock, local_sock)).start()
    
    # 启动本地端口监听
    server_sock = socket(AF_INET, SOCK_STREAM)
    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_sock.bind(('localhost', local_port))
    server_sock.listen(5)
    print(f"本地端口 {local_port} 已转发到目标实例端口 {remote_port}")
    
    while True:
        local_sock, addr = server_sock.accept()
        threading.Thread(target=handle_local_connection, args=(local_sock,)).start()

# 启动端口转发（示例：本地8080转发到目标实例80端口）
threading.Thread(
    target=forward_local_port,
    args=(8080, 80, transport)
).start()

5. 资源清理

捕获中断信号，关闭SSH连接和WebSocket通道：

try:
    while True:
        pass  # 保持主进程运行
except KeyboardInterrupt:
    print("正在关闭连接...")
    transport.close()
    ssm_socket.close()

关键注意事项 #

• IAM权限：执行代码的AWS身份需具备ssm:StartSession权限，目标实例需关联包含SSM访问权限的IAM角色，且SSM Agent已安装并正常运行。
• 依赖安装：需要提前安装依赖包：

  pip install boto3 websocket-client paramiko
  

• 异步优化：如果需要更高性能，可使用websockets（异步WebSocket库）和asyncssh（异步SSH库）替代同步实现。

内容的提问来源于stack exchange，提问作者Shuzheng