codekingpro/portable-devtools
114k
1# This file is part of paramiko.2#3# Paramiko is free software; you can redistribute it and/or modify it under the4# terms of the GNU Lesser General Public License as published by the Free5# Software Foundation; either version 2.1 of the License, or (at your option)6# any later version.7#8# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY9# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR10# A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more11# details.12#13# You should have received a copy of the GNU Lesser General Public License14# along with Paramiko; if not, write to the Free Software Foundation, Inc.,15# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.16 17import bcrypt18 19from cryptography.hazmat.backends import default_backend20from cryptography.hazmat.primitives.ciphers import Cipher21 22import nacl.signing23 24from paramiko.message import Message25from paramiko.pkey import PKey, OPENSSH_AUTH_MAGIC, _unpad_openssh26from paramiko.util import b27from paramiko.ssh_exception import SSHException, PasswordRequiredException28 29 30class Ed25519Key(PKey):31 """32 Representation of an `Ed25519 <https://ed25519.cr.yp.to/>`_ key.33 34 .. note::35 Ed25519 key support was added to OpenSSH in version 6.5.36 37 .. versionadded:: 2.238 .. versionchanged:: 2.339 Added a ``file_obj`` parameter to match other key classes.40 """41 42 name = "ssh-ed25519"43 44 def __init__(45 self, msg=None, data=None, filename=None, password=None, file_obj=None46 ):47 self.public_blob = None48 verifying_key = signing_key = None49 if msg is None and data is not None:50 msg = Message(data)51 if msg is not None:52 self._check_type_and_load_cert(53 msg=msg,54 key_type=self.name,55 cert_type="ssh-ed25519-cert-v01@openssh.com",56 )57 verifying_key = nacl.signing.VerifyKey(msg.get_binary())58 elif filename is not None:59 with open(filename, "r") as f:60 pkformat, data = self._read_private_key("OPENSSH", f)61 elif file_obj is not None:62 pkformat, data = self._read_private_key("OPENSSH", file_obj)63 64 if filename or file_obj:65 signing_key = self._parse_signing_key_data(data, password)66 67 if signing_key is None and verifying_key is None:68 raise ValueError("need a key")69 70 self._signing_key = signing_key71 self._verifying_key = verifying_key72 73 def _parse_signing_key_data(self, data, password):74 from paramiko.transport import Transport75 76 # We may eventually want this to be usable for other key types, as77 # OpenSSH moves to it, but for now this is just for Ed25519 keys.78 # This format is described here:79 # https://github.com/openssh/openssh-portable/blob/master/PROTOCOL.key80 # The description isn't totally complete, and I had to refer to the81 # source for a full implementation.82 message = Message(data)83 if message.get_bytes(len(OPENSSH_AUTH_MAGIC)) != OPENSSH_AUTH_MAGIC:84 raise SSHException("Invalid key")85 86 ciphername = message.get_text()87 kdfname = message.get_text()88 kdfoptions = message.get_binary()89 num_keys = message.get_int()90 91 if kdfname == "none":92 # kdfname of "none" must have an empty kdfoptions, the ciphername93 # must be "none"94 if kdfoptions or ciphername != "none":95 raise SSHException("Invalid key")96 elif kdfname == "bcrypt":97 if not password:98 raise PasswordRequiredException(99 "Private key file is encrypted"100 )101 kdf = Message(kdfoptions)102 bcrypt_salt = kdf.get_binary()103 bcrypt_rounds = kdf.get_int()104 else:105 raise SSHException("Invalid key")106 107 if ciphername != "none" and ciphername not in Transport._cipher_info:108 raise SSHException("Invalid key")109 110 public_keys = []111 for _ in range(num_keys):112 pubkey = Message(message.get_binary())113 if pubkey.get_text() != self.name:114 raise SSHException("Invalid key")115 public_keys.append(pubkey.get_binary())116 117 private_ciphertext = message.get_binary()118 if ciphername == "none":119 private_data = private_ciphertext120 else:121 cipher = Transport._cipher_info[ciphername]122 key = bcrypt.kdf(123 password=b(password),124 salt=bcrypt_salt,125 desired_key_bytes=cipher["key-size"] + cipher["block-size"],126 rounds=bcrypt_rounds,127 # We can't control how many rounds are on disk, so no sense128 # warning about it.129 ignore_few_rounds=True,130 )131 decryptor = Cipher(132 cipher["class"](key[: cipher["key-size"]]),133 cipher["mode"](key[cipher["key-size"] :]),134 backend=default_backend(),135 ).decryptor()136 private_data = (137 decryptor.update(private_ciphertext) + decryptor.finalize()138 )139 140 message = Message(_unpad_openssh(private_data))141 if message.get_int() != message.get_int():142 raise SSHException("Invalid key")143 144 signing_keys = []145 for i in range(num_keys):146 if message.get_text() != self.name:147 raise SSHException("Invalid key")148 # A copy of the public key, again, ignore.149 public = message.get_binary()150 key_data = message.get_binary()151 # The second half of the key data is yet another copy of the public152 # key...153 signing_key = nacl.signing.SigningKey(key_data[:32])154 # Verify that all the public keys are the same...155 assert (156 signing_key.verify_key.encode()157 == public158 == public_keys[i]159 == key_data[32:]160 )161 signing_keys.append(signing_key)162 # Comment, ignore.163 message.get_binary()164 165 if len(signing_keys) != 1:166 raise SSHException("Invalid key")167 return signing_keys[0]168 169 def asbytes(self):170 if self.can_sign():171 v = self._signing_key.verify_key172 else:173 v = self._verifying_key174 m = Message()175 m.add_string(self.name)176 m.add_string(v.encode())177 return m.asbytes()178 179 @property180 def _fields(self):181 if self.can_sign():182 v = self._signing_key.verify_key183 else:184 v = self._verifying_key185 return (self.get_name(), v)186 187 # TODO 4.0: remove188 def get_name(self):189 return self.name190 191 def get_bits(self):192 return 256193 194 def can_sign(self):195 return self._signing_key is not None196 197 def sign_ssh_data(self, data, algorithm=None):198 m = Message()199 m.add_string(self.name)200 m.add_string(self._signing_key.sign(data).signature)201 return m202 203 def verify_ssh_sig(self, data, msg):204 if msg.get_text() != self.name:205 return False206 207 try:208 self._verifying_key.verify(data, msg.get_binary())209 except nacl.exceptions.BadSignatureError:210 return False211 else:212 return True213 