如何获取Paramiko中Channel实例用于身份认证的凭据？

Paramiko的Channel实例本身没有直接提供获取认证凭据的方法，因为认证逻辑是在Transport和ServerInterface层面处理的。可以通过改造自定义的ServerInterface实现类，在认证成功时保存凭据，并将其与对应的连接绑定，从而在后续获取Channel后拿到关联的凭据。

具体实现步骤 #

• 为自定义Server类添加属性存储认证成功的用户名和密码
• 在check_auth_password方法中验证通过后，将凭据存入当前Server实例
• 为每个连接创建独立的Server实例（避免多线程下的资源冲突）
• 获取Channel后，通过transport.server拿到对应的Server实例，进而提取凭据

修改后的完整代码 #

import paramiko
import socket  
import threading  

class Server(paramiko.server.ServerInterface): 
    def __init__(self):
        # 初始化存储凭据的属性
        self.username = None
        self.password = None

    def get_allowed_auths(self, username): 
        return "password"
    
    def check_channel_request(self, kind, channelID):
        return paramiko.OPEN_SUCCEEDED
    
    def check_channel_shell_request(self, channel):
        return True
    
    def check_channel_pty_request(self, c, t, w, h, p, ph, m):
        return True
    
    def get_banner(self):
        return ("Paramiko SSH Server v1.0\n\r", "EN")

    def check_auth_password(self, username, password):
        print(f"[*] Auth request with credentials {username}:{password}")
        # 验证成功后保存凭据到当前实例
        self.username = username
        self.password = password
        return paramiko.AUTH_SUCCESSFUL

def handle(conn, addr):
    print("[*] Handler waiting for SSH connection...")
    transport = paramiko.Transport(conn)
    transport.add_server_key(host_key)
    # 为当前连接创建独立的Server实例，避免多线程冲突
    server = Server()
    transport.start_server(server=server)
    channel = transport.accept(30) 
    if channel:
        print("[*] SSH connection received")
        # 从transport关联的server实例中获取凭据
        username = server.username
        password = server.password
        print(f"[*] Channel authenticated with: {username}:{password}")
        channel.send(f"Hi {username} :)\r\n")
        print(f"[>] {channel.recv(1024).decode('utf-8').strip()}")
        channel.close()


host_key = paramiko.RSAKey.generate(2048)
sock = socket.socket()
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(("127.0.0.1", 5555))
sock.listen(100)
print("[*] Socket listening")
while True:
    conn, addr = sock.accept()
    print(f"[*] Connection from {addr[0]}:{addr[1]}, starting handler...")
    handler = threading.Thread(target=handle, args=(conn, addr))
    handler.start()
    print("[*] Started handler...")

关键说明 #

• 每个连接对应独立的Server实例：原来的全局server变量会导致多线程下不同连接的凭据互相覆盖，改为在handle函数内创建实例，确保每个连接的凭据独立存储
• 通过transport.server关联实例：Transport启动时传入的server实例会被保留，后续可以直接通过该属性获取，进而拿到存储的认证凭据
• 凭据仅在认证成功后有效：如果认证失败，username和password会保持None，使用时可以添加判断逻辑

内容的提问来源于stack exchange，提问作者TheTridentGuy supports Ukraine