获取用于在 Paramiko 中进行身份验证的“Channel”实例的凭据

Question

我一直在使用我制作的多线程 SSH 服务器，但遇到了问题。我需要一种方法来获取每个通道用于身份验证的凭据。例如，在

handle

中，在我接受来自

channel

的

transport

后，我想知道该通道使用什么凭据向服务器验证自身身份，例如

channel.get_credentials() -> (username, password)

。阅读文档，我找不到方法，但希望有人知道我不知道的东西。

import paramiko
import socket  
import threading  

class Server(paramiko.server.ServerInterface): 
    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}")
        return paramiko.AUTH_SUCCESSFUL

def handle(conn, addr):
    print("[*] Handler waiting for SSH connection...")
    transport = paramiko.Transport(conn)
    transport.add_server_key(host_key)
    transport.start_server(server=server)
    channel = transport.accept(30) 
    if channel:
        print("[*] SSH connection recieved")
        channel.send("Hi :)\r\n")
        print(f"[>] {channel.recv(1024)}")
        channel.close()


host_key = paramiko.RSAKey.generate(2048)
server = Server()
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...")

Answer 1

我相信您必须记住自己代码中的凭据。我怀疑帕拉米科还记得你。无论如何会怎么样？ SSH 身份验证可能很复杂。从理论上讲，这可能需要多个因素。

Imo，Paramiko 服务器 API 的想法是没有单个全局

ServerInterface

，但每个连接都有自己的实例。在这种情况下，您可以存储登录身份。

class Server(paramiko.server.ServerInterface): 
    def check_auth_password(self, username, password):
        print(f"[*] Auth request with credentials {username}:{password}")
        self.username = username
        return paramiko.AUTH_SUCCESSFUL

server = Server()
transport.start_server(server=server)
channel = transport.accept(30)
print(f"Connection by {server.username}...")