如何使用 paramiko 的 Pkey._read_private_key() 获取密钥并从 OPENSSH 格式解析它们？还将它们传递给 paramiko 并验证它们

Question

我的问题是我想使用：https://github.com/paramiko/paramiko/blob/main/paramiko/pkey.py#L498 解析密钥，无论其格式如何，因为它们是通过以下方式创建的：

 ssh-keygen -f <file> -t <algo>

它们有一个标题：“BEGIN OPENSSH PRIVATE KEY” 所以我不太确定我应该如何弄清楚它是什么格式。

我尝试过：https://github.com/paramiko/paramiko/blob/main/paramiko/pkey.py#L473 tag：

"RSA"

或

"DSA"

，用于标记数据块的标签。 但我感觉自己没有任何进步。最好应该是 pkey._read_private_key() 方法

class ParaProxy(paramiko.proxy.ProxyCommand):
    def __init__(self, stdin, stdout, stderr):
        self.stdin = stdin
        self.stdout = stdout
        self.stderr = stderr
        self.timeout = None
        self.channel = stdin.channel

    def send(self, content):
        try:
            self.stdin.write(content)
        except IOError as exc:
            raise socket.error("Error: {}".format(exc))
        return len(content)

    def recv(self, size):
        try:
            buffer = b''
            start = time.time()

            while len(buffer) < size:
                select_timeout = self._calculate_remaining_time(start)
                ready, _, _ = select([self.stdout.channel], [], [],
                                     select_timeout)
                if ready and self.stdout.channel is ready[0]:
                    buffer += self.stdout.read(size - len(buffer))

        except socket.timeout:
            if not buffer:
                raise

        except IOError as e:
            return ""

        return buffer

    def _calculate_remaining_time(self, start):
        if self.timeout is not None:
            elapsed = time.time() - start
            if elapsed >= self.timeout:
                raise socket.timeout()
            return self.timeout - elapsed
        return None

    def close(self):
        self.stdin.close()
        self.stdout.close()
        self.stderr.close()
        self.channel.close()

host = '<IP>'
port = 22
username = '<USERNAME>'
password = '<PASSWORD>'

key_passphrase = '11223344'
rsa_key_filename = '/home/USER/.ssh/try_with_passphrase'
rsa_key_with_passphrase = paramiko.RSAKey.from_private_key_file(rsa_key_filename, key_passphrase)
dsa_key_filename = '/home/USER/.ssh/id_dsa'
dsa_key_with_passphrase = paramiko.DSSKey.from_private_key_file(dsa_key_filename, key_passphrase)
ecdsa_key_filename = '/home/USER/.ssh/id_ecdsa'
ecdsa_key_with_passphrase = paramiko.ECDSAKey.from_private_key_file(ecdsa_key_filename, key_passphrase)

mid_cli = paramiko.SSHClient()
mid_cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
mid_cli.connect(hostname=host, username=username, password=password, pkey=dsa_key_with_passphrase)
io_tupple = mid_cli.exec_command('nc {} {}'.format(host, port))

proxy = ParaProxy(*io_tupple)

end_cli = paramiko.SSHClient()
end_cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())

end_cli.connect(hostname=host, username=username, password=password, sock=proxy, pkey=dsa_key_with_passphrase)

commands = ['ls', 'echo $USER', 'hostname', 'echo close', 'bad_command']
for command in commands:
    print("Executing command:  ", command)
    stdin, stdout, stderr = end_cli.exec_command(command)
    output = stdout.read().decode()
    err = stderr.read().decode()

    if err:
        print('Error:', err)
    else:
        print("Output:")
        print(output)

proxy.close()
time.sleep(.1)
mid_cli.close()
end_cli.close()

我已经尝试过：

content = """
-----BEGIN OPENSSH PRIVATE KEY-----
b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAABlwAAAAdzc2gtcn
NhAAAAAwEAAQAAAYEAtY6YhI1U/SrSN9ieN1XR42q9vtZ3xA6OQecRzsFwtXyfS7YdiWL9
WILlDMuCFFw4L7g7iB5QtOmaChNwCO4Xtuotno6hmFuVELtamL0Dk6JjnnsLrsgRhtbrud
JsEsNe9B/j6vCeoAo4mzWb8oGwqyG0PfMyiQ9EFPm3FLAIYfDxnG5dUnqesHMwSDrG9STG
aAswMv+nm6ymTEIGP3gMtDkeiTDWaMkOxl9eJ4IqRDXSlnIrGhuEF0lqxptAe3ObEE6esl
u1WtYWrrhonE38M1R4xdXoIhBlyaQKWO9ABudIW8dBYpEpqWW19rt1svu3AKsrMPdBpSd9
YlOkpTinnXne1dOg6wS+l7ztkoLUhPrILlpfYM6M1Qt2JmCtGTP+MGz78DvbNIG0NZqcV6
6UFydHeK04hJ62QhHLwqcT+5I/OXz7WXEHF7yexLfy4W2E2+5JDJqrG7AuHLnPYkgkCSX2
PWPjvINzPD3Ep3SmyPpnNKRRzZvb8KRS4zCV3P+pAAAFkDsRo2Q7EaNkAAAAB3NzaC1yc2
EAAAGBALWOmISNVP0q0jfYnjdV0eNqvb7Wd8QOjkHnEc7BcLV8n0u2HYli/ViC5QzLghRc
OC+4O4geULTpmgoTcAjuF7bqLZ6OoZhblRC7Wpi9A5OiY557C67IEYbW67nSbBLDXvQf4+
rwnqAKOJs1m/KBsKshtD3zMokPRBT5txSwCGHw8ZxuXVJ6nrBzMEg6xvUkxmgLMDL/p5us
pkxCBj94DLQ5Hokw1mjJDsZfXieCKkQ10pZyKxobhBdJasabQHtzmxBOnrJbtVrWFq64aJ
xN/DNUeMXV6CIQZcmkCljvQAbnSFvHQWKRKalltfa7dbL7twCrKzD3QaUnfWJTpKU4p515
3tXToOsEvpe87ZKC1IT6yC5aX2DOjNULdiZgrRkz/jBs+/A72zSBtDWanFeulBcnR3itOI
SetkIRy8KnE/uSPzl8+1lxBxe8nsS38uFthNvuSQyaqxuwLhy5z2JIJAkl9j1j47yDczw9
xKd0psj6ZzSkUc2b2/CkUuMwldz/qQAAAAMBAAEAAAGAD4wZHTO9oQrlUtU00kQB+92oh0
K2IZbGAZh8HsaTpGT6PqlnHNxaWdWmr+pu9pehk5uNFIZVDrL45w88TzV1WCt0UeFArwt7
p8HFe7Im7+HucjQy/0oAbzmxGLnvWYl7tH/VHImzqbyKVac49iK7OmJh3RK6A4DoSXwLXK
UI6Lsxc34DdZRMycspaawH3NcD+wWnQFzCcXsux/W4838wJlBEYEbre3BpOJ1S0qMLw8jQ
omO/M2aDEJvvvfduRYwUCJaOTJJNnsvPPiwG0pBmFo0IRPcomogHJelhmHoCdiK33CoJhY
JVLWyHZHW7BSQHzbyCC4gdqEvZWSgivMkxzybOdUMcBe83OnCMf69llTQxOn15IzCMS2pq
9wc4+juNiuKpOzqiBrksw93tjQcTL9u0gHswaEpVFAjSJr5UcYWzEEX6hpBfzVWX35p3Ca
/3I7M6kqkzyV7o830nUYVE+A/6tp6jlMcpRTpDyB+tn9SCbwIy+zf1fBxmLHpjo/wRAAAA
wEcIlgAU0Kzz2jyBfcmxWjQrMUW4mstmWPIqKjKg9BcDM5u06FchsjFAyBMnZ7zergs78j
/SL2lLfYnzwjFYkgb+LYVUmk4qZ4/W2/MGpqVbEG7MTDM3RQ+t2SZXgIQFM7E75kSjY+pD
QYNRWw9RXIj2PFt2HV+HtDtF2X0a1dJpqr3JNkKHNnZpqB8ZAAGil2RAjfnQaKpr1GgPNX
scmrX+1YMQq3sEXB9hXYd6t8kcZDMblTdtdiiNYrZeVY05FQAAAMEAxSSKux0zHj2bruBM
02yfN68CfaUZFaT4CnXXEH3NKEthuVsfibPtlTiROvjSe2130cKzkzfiLhIOWLATHV8IIl
+fhJXMv01yO3HoW1QcQhuGc9iH49cRHF1u4vtP9VinHdR7HnkZU8/OjJvVQggfV4YcX0y/
CwKwbwHlaDoXKNGqcbEwitsNYFDGW2rNrXz5ln426h6DgCrHCg41t5OLmYa37T738yukHL
GEetzKQ9qh+44EjVmwrhExqYa75rhxAAAAwQDrwtl5pgkrKBZ2sFZ3BNVwU0nF0yVDu96C
2M7IB9oC00FroDArFNfICN1eVBJ5MyTMse619768uKI/G1TGPPU0hMWMnlI0Vn0wx8JCJu
+QwPWwyCnlhTv2xpBFnepYTrwembTpX3x/Uzpzp/sVJIiDt+/Xfl6Be/vdEWKfwx5HD0OX
fjWaNOlntoIsA0sFybjCbLMvccBGC7bZAezN98sVJtcOzQ/lHJ7oMRrQgMTFY+hv9lTZiP
jbgBbkquDIFrkAAAAXbWVyZXRyaWN1bGFAbWVyZXRyaWN1bGEBAgME
-----END OPENSSH PRIVATE KEY-----
"""
file = StringIO(content)
lines = file.readlines()

# Load the RSA private key
key = paramiko.pkey.PKey()
method_key = key._read_private_key_file(key, filename=dsa_key_filename, password=key_passphrase)
method_key = key._read_private_key(key, f=file, password=key_passphrase)
key._read_private_key_openssh(key, lines)

Answer 1

def get_key_class_name(filepath, passphrase):
    for pkey_class in (RSAKey, DSSKey, ECDSAKey, Ed25519Key):
        try:
            key = pkey_class.from_private_key_file(filepath, passphrase)
            class_name = str(pkey_class).split('.')[-1].rstrip(">'")

            return class_name

        except Exception as e:
            # print("An exception occurred: {}".format(e))
            pass

def proper_method_call(filepath, passphrase):
    key_class_name = get_key_class_name(filepath, passphrase)
    key_class = getattr(paramiko, key_class_name)
    key_method = getattr(key_class, "from_private_key_file")
    key = key_method(filepath, passphrase)
    return key

rsa_key_with_passphrase = proper_method_call(rsa_key_filename, key_passphrase)

将其作为参数传递：

pkey=rsa_key_with_passphrase

用途：

mid_cli = paramiko.SSHClient()
mid_cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
mid_cli.connect(hostname=host, username=username, password=password, pkey=rsa_key_with_passphrase)
io_tupple = mid_cli.exec_command('nc {} {}'.format(host, port))

proxy = ParaProxy(*io_tupple)

end_cli = paramiko.SSHClient()
end_cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())

end_cli.connect(hostname=host, username=username, password=password, sock=proxy, pkey=rsa_key_with_passphrase)

commands = ['ls', 'echo $USER', 'hostname', 'echo close', 'bad_command']
for command in commands:
    print("Executing command:  ", command)
    stdin, stdout, stderr = end_cli.exec_command(command)
    output = stdout.read().decode()
    err = stderr.read().decode()
    if err:
        print('Error:', err)
    else:
        print("Output:")
        print(output)

proxy.close()
time.sleep(.1)
mid_cli.close()
end_cli.close()