codekingpro/portable-devtools
114k
1# Copyright (C) 2003-2007 Robey Pointer <robeypointer@gmail.com>2#3# This file is part of paramiko.4#5# Paramiko is free software; you can redistribute it and/or modify it under the6# terms of the GNU Lesser General Public License as published by the Free7# Software Foundation; either version 2.1 of the License, or (at your option)8# any later version.9#10# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY11# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR12# A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more13# details.14#15# You should have received a copy of the GNU Lesser General Public License16# along with Paramiko; if not, write to the Free Software Foundation, Inc.,17# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.18 19"""20Common API for all public keys.21"""22 23import base6424from base64 import encodebytes, decodebytes25from binascii import unhexlify26import os27from pathlib import Path28from hashlib import md5, sha25629import re30import struct31 32import bcrypt33 34from cryptography.hazmat.backends import default_backend35from cryptography.hazmat.primitives import serialization36from cryptography.hazmat.primitives.ciphers import algorithms, modes, Cipher37from cryptography.hazmat.primitives import asymmetric38 39from paramiko import util40from paramiko.util import u, b41from paramiko.common import o60042from paramiko.ssh_exception import SSHException, PasswordRequiredException43from paramiko.message import Message44 45 46OPENSSH_AUTH_MAGIC = b"openssh-key-v1\x00"47 48 49def _unpad_openssh(data):50 # At the moment, this is only used for unpadding private keys on disk. This51 # really ought to be made constant time (possibly by upstreaming this logic52 # into pyca/cryptography).53 padding_length = data[-1]54 if 0x20 <= padding_length < 0x7F:55 return data # no padding, last byte part comment (printable ascii)56 if padding_length > 15:57 raise SSHException("Invalid key")58 for i in range(padding_length):59 if data[i - padding_length] != i + 1:60 raise SSHException("Invalid key")61 return data[:-padding_length]62 63 64class UnknownKeyType(Exception):65 """66 An unknown public/private key algorithm was attempted to be read.67 """68 69 def __init__(self, key_type=None, key_bytes=None):70 self.key_type = key_type71 self.key_bytes = key_bytes72 73 def __str__(self):74 return f"UnknownKeyType(type={self.key_type!r}, bytes=<{len(self.key_bytes)}>)" # noqa75 76 77class PKey:78 """79 Base class for public keys.80 81 Also includes some "meta" level convenience constructors such as82 `.from_type_string`.83 """84 85 # known encryption types for private key files:86 _CIPHER_TABLE = {87 "AES-128-CBC": {88 "cipher": algorithms.AES,89 "keysize": 16,90 "blocksize": 16,91 "mode": modes.CBC,92 },93 "AES-256-CBC": {94 "cipher": algorithms.AES,95 "keysize": 32,96 "blocksize": 16,97 "mode": modes.CBC,98 },99 "DES-EDE3-CBC": {100 "cipher": algorithms.TripleDES,101 "keysize": 24,102 "blocksize": 8,103 "mode": modes.CBC,104 },105 }106 _PRIVATE_KEY_FORMAT_ORIGINAL = 1107 _PRIVATE_KEY_FORMAT_OPENSSH = 2108 BEGIN_TAG = re.compile(109 r"^-{5}BEGIN (RSA|DSA|EC|OPENSSH) PRIVATE KEY-{5}\s*$"110 )111 END_TAG = re.compile(r"^-{5}END (RSA|DSA|EC|OPENSSH) PRIVATE KEY-{5}\s*$")112 113 @staticmethod114 def from_path(path, passphrase=None):115 """116 Attempt to instantiate appropriate key subclass from given file path.117 118 :param Path path: The path to load (may also be a `str`).119 120 :returns:121 A `PKey` subclass instance.122 123 :raises:124 `UnknownKeyType`, if our crypto backend doesn't know this key type.125 126 .. versionadded:: 3.2127 """128 # TODO: make sure sphinx is reading Path right in param list...129 130 # Lazy import to avoid circular import issues131 from paramiko import DSSKey, RSAKey, Ed25519Key, ECDSAKey132 133 # Normalize to string, as cert suffix isn't quite an extension, so134 # pathlib isn't useful for this.135 path = str(path)136 137 # Sort out cert vs key, i.e. it is 'legal' to hand this kind of API138 # /either/ the key /or/ the cert, when there is a key/cert pair.139 cert_suffix = "-cert.pub"140 if str(path).endswith(cert_suffix):141 key_path = path[: -len(cert_suffix)]142 cert_path = path143 else:144 key_path = path145 cert_path = path + cert_suffix146 147 key_path = Path(key_path).expanduser()148 cert_path = Path(cert_path).expanduser()149 150 data = key_path.read_bytes()151 # Like OpenSSH, try modern/OpenSSH-specific key load first152 try:153 loaded = serialization.load_ssh_private_key(154 data=data, password=passphrase155 )156 # Then fall back to assuming legacy PEM type157 except ValueError:158 loaded = serialization.load_pem_private_key(159 data=data, password=passphrase160 )161 # TODO Python 3.10: match statement? (NOTE: we cannot use a dict162 # because the results from the loader are literal backend, eg openssl,163 # private classes, so isinstance tests work but exact 'x class is y'164 # tests will not work)165 # TODO: leverage already-parsed/math'd obj to avoid duplicate cpu166 # cycles? seemingly requires most of our key subclasses to be rewritten167 # to be cryptography-object-forward. this is still likely faster than168 # the old SSHClient code that just tried instantiating every class!169 key_class = None170 if isinstance(loaded, asymmetric.dsa.DSAPrivateKey):171 key_class = DSSKey172 elif isinstance(loaded, asymmetric.rsa.RSAPrivateKey):173 key_class = RSAKey174 elif isinstance(loaded, asymmetric.ed25519.Ed25519PrivateKey):175 key_class = Ed25519Key176 elif isinstance(loaded, asymmetric.ec.EllipticCurvePrivateKey):177 key_class = ECDSAKey178 else:179 raise UnknownKeyType(key_bytes=data, key_type=loaded.__class__)180 with key_path.open() as fd:181 key = key_class.from_private_key(fd, password=passphrase)182 if cert_path.exists():183 # load_certificate can take Message, path-str, or value-str184 key.load_certificate(str(cert_path))185 return key186 187 @staticmethod188 def from_type_string(key_type, key_bytes):189 """190 Given type `str` & raw `bytes`, return a `PKey` subclass instance.191 192 For example, ``PKey.from_type_string("ssh-ed25519", <public bytes>)``193 will (if successful) return a new `.Ed25519Key`.194 195 :param str key_type:196 The key type, eg ``"ssh-ed25519"``.197 :param bytes key_bytes:198 The raw byte data forming the key material, as expected by199 subclasses' ``data`` parameter.200 201 :returns:202 A `PKey` subclass instance.203 204 :raises:205 `UnknownKeyType`, if no registered classes knew about this type.206 207 .. versionadded:: 3.2208 """209 from paramiko import key_classes210 211 for key_class in key_classes:212 if key_type in key_class.identifiers():213 # TODO: needs to passthru things like passphrase214 return key_class(data=key_bytes)215 raise UnknownKeyType(key_type=key_type, key_bytes=key_bytes)216 217 @classmethod218 def identifiers(cls):219 """220 returns an iterable of key format/name strings this class can handle.221 222 Most classes only have a single identifier, and thus this default223 implementation suffices; see `.ECDSAKey` for one example of an224 override.225 """226 return [cls.name]227 228 # TODO 4.0: make this and subclasses consistent, some of our own229 # classmethods even assume kwargs we don't define!230 # TODO 4.0: prob also raise NotImplementedError instead of pass'ing; the231 # contract is pretty obviously that you need to handle msg/data/filename232 # appropriately. (If 'pass' is a concession to testing, see about doing the233 # work to fix the tests instead)234 def __init__(self, msg=None, data=None):235 """236 Create a new instance of this public key type. If ``msg`` is given,237 the key's public part(s) will be filled in from the message. If238 ``data`` is given, the key's public part(s) will be filled in from239 the string.240 241 :param .Message msg:242 an optional SSH `.Message` containing a public key of this type.243 :param bytes data:244 optional, the bytes of a public key of this type245 246 :raises: `.SSHException` --247 if a key cannot be created from the ``data`` or ``msg`` given, or248 no key was passed in.249 """250 pass251 252 # TODO: arguably this might want to be __str__ instead? ehh253 # TODO: ditto the interplay between showing class name (currently we just254 # say PKey writ large) and algorithm (usually == class name, but not255 # always, also sometimes shows certificate-ness)256 # TODO: if we do change it, we also want to tweak eg AgentKey, as it257 # currently displays agent-ness with a suffix258 def __repr__(self):259 comment = ""260 # Works for AgentKey, may work for others?261 if hasattr(self, "comment") and self.comment:262 comment = f", comment={self.comment!r}"263 return f"PKey(alg={self.algorithm_name}, bits={self.get_bits()}, fp={self.fingerprint}{comment})" # noqa264 265 # TODO 4.0: just merge into __bytes__ (everywhere)266 def asbytes(self):267 """268 Return a string of an SSH `.Message` made up of the public part(s) of269 this key. This string is suitable for passing to `__init__` to270 re-create the key object later.271 """272 return bytes()273 274 def __bytes__(self):275 return self.asbytes()276 277 def __eq__(self, other):278 return isinstance(other, PKey) and self._fields == other._fields279 280 def __hash__(self):281 return hash(self._fields)282 283 @property284 def _fields(self):285 raise NotImplementedError286 287 def get_name(self):288 """289 Return the name of this private key implementation.290 291 :return:292 name of this private key type, in SSH terminology, as a `str` (for293 example, ``"ssh-rsa"``).294 """295 return ""296 297 @property298 def algorithm_name(self):299 """300 Return the key algorithm identifier for this key.301 302 Similar to `get_name`, but aimed at pure algorithm name instead of SSH303 protocol field value.304 """305 # Nuke the leading 'ssh-'306 # TODO in Python 3.9: use .removeprefix()307 name = self.get_name().replace("ssh-", "")308 # Trim any cert suffix (but leave the -cert, as OpenSSH does)309 cert_tail = "-cert-v01@openssh.com"310 if cert_tail in name:311 name = name.replace(cert_tail, "-cert")312 # Nuke any eg ECDSA suffix, OpenSSH does basically this too.313 else:314 name = name.split("-")[0]315 return name.upper()316 317 def get_bits(self):318 """319 Return the number of significant bits in this key. This is useful320 for judging the relative security of a key.321 322 :return: bits in the key (as an `int`)323 """324 # TODO 4.0: raise NotImplementedError, 0 is unlikely to ever be325 # _correct_ and nothing in the critical path seems to use this.326 return 0327 328 def can_sign(self):329 """330 Return ``True`` if this key has the private part necessary for signing331 data.332 """333 return False334 335 def get_fingerprint(self):336 """337 Return an MD5 fingerprint of the public part of this key. Nothing338 secret is revealed.339 340 :return:341 a 16-byte `string <str>` (binary) of the MD5 fingerprint, in SSH342 format.343 """344 return md5(self.asbytes()).digest()345 346 @property347 def fingerprint(self):348 """349 Modern fingerprint property designed to be comparable to OpenSSH.350 351 Currently only does SHA256 (the OpenSSH default).352 353 .. versionadded:: 3.2354 """355 hashy = sha256(bytes(self))356 hash_name = hashy.name.upper()357 b64ed = encodebytes(hashy.digest())358 cleaned = u(b64ed).strip().rstrip("=") # yes, OpenSSH does this too!359 return f"{hash_name}:{cleaned}"360 361 def get_base64(self):362 """363 Return a base64 string containing the public part of this key. Nothing364 secret is revealed. This format is compatible with that used to store365 public key files or recognized host keys.366 367 :return: a base64 `string <str>` containing the public part of the key.368 """369 return u(encodebytes(self.asbytes())).replace("\n", "")370 371 def sign_ssh_data(self, data, algorithm=None):372 """373 Sign a blob of data with this private key, and return a `.Message`374 representing an SSH signature message.375 376 :param bytes data:377 the data to sign.378 :param str algorithm:379 the signature algorithm to use, if different from the key's380 internal name. Default: ``None``.381 :return: an SSH signature `message <.Message>`.382 383 .. versionchanged:: 2.9384 Added the ``algorithm`` kwarg.385 """386 return bytes()387 388 def verify_ssh_sig(self, data, msg):389 """390 Given a blob of data, and an SSH message representing a signature of391 that data, verify that it was signed with this key.392 393 :param bytes data: the data that was signed.394 :param .Message msg: an SSH signature message395 :return:396 ``True`` if the signature verifies correctly; ``False`` otherwise.397 """398 return False399 400 @classmethod401 def from_private_key_file(cls, filename, password=None):402 """403 Create a key object by reading a private key file. If the private404 key is encrypted and ``password`` is not ``None``, the given password405 will be used to decrypt the key (otherwise `.PasswordRequiredException`406 is thrown). Through the magic of Python, this factory method will407 exist in all subclasses of PKey (such as `.RSAKey` or `.DSSKey`), but408 is useless on the abstract PKey class.409 410 :param str filename: name of the file to read411 :param str password:412 an optional password to use to decrypt the key file, if it's413 encrypted414 :return: a new `.PKey` based on the given private key415 416 :raises: ``IOError`` -- if there was an error reading the file417 :raises: `.PasswordRequiredException` -- if the private key file is418 encrypted, and ``password`` is ``None``419 :raises: `.SSHException` -- if the key file is invalid420 """421 key = cls(filename=filename, password=password)422 return key423 424 @classmethod425 def from_private_key(cls, file_obj, password=None):426 """427 Create a key object by reading a private key from a file (or file-like)428 object. If the private key is encrypted and ``password`` is not429 ``None``, the given password will be used to decrypt the key (otherwise430 `.PasswordRequiredException` is thrown).431 432 :param file_obj: the file-like object to read from433 :param str password:434 an optional password to use to decrypt the key, if it's encrypted435 :return: a new `.PKey` based on the given private key436 437 :raises: ``IOError`` -- if there was an error reading the key438 :raises: `.PasswordRequiredException` --439 if the private key file is encrypted, and ``password`` is ``None``440 :raises: `.SSHException` -- if the key file is invalid441 """442 key = cls(file_obj=file_obj, password=password)443 return key444 445 def write_private_key_file(self, filename, password=None):446 """447 Write private key contents into a file. If the password is not448 ``None``, the key is encrypted before writing.449 450 :param str filename: name of the file to write451 :param str password:452 an optional password to use to encrypt the key file453 454 :raises: ``IOError`` -- if there was an error writing the file455 :raises: `.SSHException` -- if the key is invalid456 """457 raise Exception("Not implemented in PKey")458 459 def write_private_key(self, file_obj, password=None):460 """461 Write private key contents into a file (or file-like) object. If the462 password is not ``None``, the key is encrypted before writing.463 464 :param file_obj: the file-like object to write into465 :param str password: an optional password to use to encrypt the key466 467 :raises: ``IOError`` -- if there was an error writing to the file468 :raises: `.SSHException` -- if the key is invalid469 """470 # TODO 4.0: NotImplementedError (plus everywhere else in here)471 raise Exception("Not implemented in PKey")472 473 def _read_private_key_file(self, tag, filename, password=None):474 """475 Read an SSH2-format private key file, looking for a string of the type476 ``"BEGIN xxx PRIVATE KEY"`` for some ``xxx``, base64-decode the text we477 find, and return it as a string. If the private key is encrypted and478 ``password`` is not ``None``, the given password will be used to479 decrypt the key (otherwise `.PasswordRequiredException` is thrown).480 481 :param str tag: ``"RSA"`` or ``"DSA"``, the tag used to mark the482 data block.483 :param str filename: name of the file to read.484 :param str password:485 an optional password to use to decrypt the key file, if it's486 encrypted.487 :return: the `bytes` that make up the private key.488 489 :raises: ``IOError`` -- if there was an error reading the file.490 :raises: `.PasswordRequiredException` -- if the private key file is491 encrypted, and ``password`` is ``None``.492 :raises: `.SSHException` -- if the key file is invalid.493 """494 with open(filename, "r") as f:495 data = self._read_private_key(tag, f, password)496 return data497 498 def _read_private_key(self, tag, f, password=None):499 lines = f.readlines()500 if not lines:501 raise SSHException("no lines in {} private key file".format(tag))502 503 # find the BEGIN tag504 start = 0505 m = self.BEGIN_TAG.match(lines[start])506 line_range = len(lines) - 1507 while start < line_range and not m:508 start += 1509 m = self.BEGIN_TAG.match(lines[start])510 start += 1511 keytype = m.group(1) if m else None512 if start >= len(lines) or keytype is None:513 raise SSHException("not a valid {} private key file".format(tag))514 515 # find the END tag516 end = start517 m = self.END_TAG.match(lines[end])518 while end < line_range and not m:519 end += 1520 m = self.END_TAG.match(lines[end])521 522 if keytype == tag:523 data = self._read_private_key_pem(lines, end, password)524 pkformat = self._PRIVATE_KEY_FORMAT_ORIGINAL525 elif keytype == "OPENSSH":526 data = self._read_private_key_openssh(lines[start:end], password)527 pkformat = self._PRIVATE_KEY_FORMAT_OPENSSH528 else:529 raise SSHException(530 "encountered {} key, expected {} key".format(keytype, tag)531 )532 533 return pkformat, data534 535 def _got_bad_key_format_id(self, id_):536 err = "{}._read_private_key() spat out an unknown key format id '{}'"537 raise SSHException(err.format(self.__class__.__name__, id_))538 539 def _read_private_key_pem(self, lines, end, password):540 start = 0541 # parse any headers first542 headers = {}543 start += 1544 while start < len(lines):545 line = lines[start].split(": ")546 if len(line) == 1:547 break548 headers[line[0].lower()] = line[1].strip()549 start += 1550 # if we trudged to the end of the file, just try to cope.551 try:552 data = decodebytes(b("".join(lines[start:end])))553 except base64.binascii.Error as e:554 raise SSHException("base64 decoding error: {}".format(e))555 if "proc-type" not in headers:556 # unencryped: done557 return data558 # encrypted keyfile: will need a password559 proc_type = headers["proc-type"]560 if proc_type != "4,ENCRYPTED":561 raise SSHException(562 'Unknown private key structure "{}"'.format(proc_type)563 )564 try:565 encryption_type, saltstr = headers["dek-info"].split(",")566 except:567 raise SSHException("Can't parse DEK-info in private key file")568 if encryption_type not in self._CIPHER_TABLE:569 raise SSHException(570 'Unknown private key cipher "{}"'.format(encryption_type)571 )572 # if no password was passed in,573 # raise an exception pointing out that we need one574 if password is None:575 raise PasswordRequiredException("Private key file is encrypted")576 cipher = self._CIPHER_TABLE[encryption_type]["cipher"]577 keysize = self._CIPHER_TABLE[encryption_type]["keysize"]578 mode = self._CIPHER_TABLE[encryption_type]["mode"]579 salt = unhexlify(b(saltstr))580 key = util.generate_key_bytes(md5, salt, password, keysize)581 decryptor = Cipher(582 cipher(key), mode(salt), backend=default_backend()583 ).decryptor()584 return decryptor.update(data) + decryptor.finalize()585 586 def _read_private_key_openssh(self, lines, password):587 """588 Read the new OpenSSH SSH2 private key format available589 since OpenSSH version 6.5590 Reference:591 https://github.com/openssh/openssh-portable/blob/master/PROTOCOL.key592 """593 try:594 data = decodebytes(b("".join(lines)))595 except base64.binascii.Error as e:596 raise SSHException("base64 decoding error: {}".format(e))597 598 # read data struct599 auth_magic = data[:15]600 if auth_magic != OPENSSH_AUTH_MAGIC:601 raise SSHException("unexpected OpenSSH key header encountered")602 603 cstruct = self._uint32_cstruct_unpack(data[15:], "sssur")604 cipher, kdfname, kdf_options, num_pubkeys, remainder = cstruct605 # For now, just support 1 key.606 if num_pubkeys > 1:607 raise SSHException(608 "unsupported: private keyfile has multiple keys"609 )610 pubkey, privkey_blob = self._uint32_cstruct_unpack(remainder, "ss")611 612 if kdfname == b("bcrypt"):613 if cipher == b("aes256-cbc"):614 mode = modes.CBC615 elif cipher == b("aes256-ctr"):616 mode = modes.CTR617 else:618 raise SSHException(619 "unknown cipher `{}` used in private key file".format(620 cipher.decode("utf-8")621 )622 )623 # Encrypted private key.624 # If no password was passed in, raise an exception pointing625 # out that we need one626 if password is None:627 raise PasswordRequiredException(628 "private key file is encrypted"629 )630 631 # Unpack salt and rounds from kdfoptions632 salt, rounds = self._uint32_cstruct_unpack(kdf_options, "su")633 634 # run bcrypt kdf to derive key and iv/nonce (32 + 16 bytes)635 key_iv = bcrypt.kdf(636 b(password),637 b(salt),638 48,639 rounds,640 # We can't control how many rounds are on disk, so no sense641 # warning about it.642 ignore_few_rounds=True,643 )644 key = key_iv[:32]645 iv = key_iv[32:]646 647 # decrypt private key blob648 decryptor = Cipher(649 algorithms.AES(key), mode(iv), default_backend()650 ).decryptor()651 decrypted_privkey = decryptor.update(privkey_blob)652 decrypted_privkey += decryptor.finalize()653 elif cipher == b("none") and kdfname == b("none"):654 # Unencrypted private key655 decrypted_privkey = privkey_blob656 else:657 raise SSHException(658 "unknown cipher or kdf used in private key file"659 )660 661 # Unpack private key and verify checkints662 cstruct = self._uint32_cstruct_unpack(decrypted_privkey, "uusr")663 checkint1, checkint2, keytype, keydata = cstruct664 665 if checkint1 != checkint2:666 raise SSHException(667 "OpenSSH private key file checkints do not match"668 )669 670 return _unpad_openssh(keydata)671 672 def _uint32_cstruct_unpack(self, data, strformat):673 """674 Used to read new OpenSSH private key format.675 Unpacks a c data structure containing a mix of 32-bit uints and676 variable length strings prefixed by 32-bit uint size field,677 according to the specified format. Returns the unpacked vars678 in a tuple.679 Format strings:680 s - denotes a string681 i - denotes a long integer, encoded as a byte string682 u - denotes a 32-bit unsigned integer683 r - the remainder of the input string, returned as a string684 """685 arr = []686 idx = 0687 try:688 for f in strformat:689 if f == "s":690 # string691 s_size = struct.unpack(">L", data[idx : idx + 4])[0]692 idx += 4693 s = data[idx : idx + s_size]694 idx += s_size695 arr.append(s)696 if f == "i":697 # long integer698 s_size = struct.unpack(">L", data[idx : idx + 4])[0]699 idx += 4700 s = data[idx : idx + s_size]701 idx += s_size702 i = util.inflate_long(s, True)703 arr.append(i)704 elif f == "u":705 # 32-bit unsigned int706 u = struct.unpack(">L", data[idx : idx + 4])[0]707 idx += 4708 arr.append(u)709 elif f == "r":710 # remainder as string711 s = data[idx:]712 arr.append(s)713 break714 except Exception as e:715 # PKey-consuming code frequently wants to save-and-skip-over issues716 # with loading keys, and uses SSHException as the (really friggin717 # awful) signal for this. So for now...we do this.718 raise SSHException(str(e))719 return tuple(arr)720 721 def _write_private_key_file(self, filename, key, format, password=None):722 """723 Write an SSH2-format private key file in a form that can be read by724 paramiko or openssh. If no password is given, the key is written in725 a trivially-encoded format (base64) which is completely insecure. If726 a password is given, DES-EDE3-CBC is used.727 728 :param str tag:729 ``"RSA"`` or ``"DSA"``, the tag used to mark the data block.730 :param filename: name of the file to write.731 :param bytes data: data blob that makes up the private key.732 :param str password: an optional password to use to encrypt the file.733 734 :raises: ``IOError`` -- if there was an error writing the file.735 """736 # Ensure that we create new key files directly with a user-only mode,737 # instead of opening, writing, then chmodding, which leaves us open to738 # CVE-2022-24302.739 with os.fdopen(740 os.open(741 filename,742 # NOTE: O_TRUNC is a noop on new files, and O_CREAT is a noop743 # on existing files, so using all 3 in both cases is fine.744 flags=os.O_WRONLY | os.O_TRUNC | os.O_CREAT,745 # Ditto the use of the 'mode' argument; it should be safe to746 # give even for existing files (though it will not act like a747 # chmod in that case).748 mode=o600,749 ),750 # Yea, you still gotta inform the FLO that it is in "write" mode.751 "w",752 ) as f:753 self._write_private_key(f, key, format, password=password)754 755 def _write_private_key(self, f, key, format, password=None):756 if password is None:757 encryption = serialization.NoEncryption()758 else:759 encryption = serialization.BestAvailableEncryption(b(password))760 761 f.write(762 key.private_bytes(763 serialization.Encoding.PEM, format, encryption764 ).decode()765 )766 767 def _check_type_and_load_cert(self, msg, key_type, cert_type):768 """769 Perform message type-checking & optional certificate loading.770 771 This includes fast-forwarding cert ``msg`` objects past the nonce, so772 that the subsequent fields are the key numbers; thus the caller may773 expect to treat the message as key material afterwards either way.774 775 The obtained key type is returned for classes which need to know what776 it was (e.g. ECDSA.)777 """778 # Normalization; most classes have a single key type and give a string,779 # but eg ECDSA is a 1:N mapping.780 key_types = key_type781 cert_types = cert_type782 if isinstance(key_type, str):783 key_types = [key_types]784 if isinstance(cert_types, str):785 cert_types = [cert_types]786 # Can't do much with no message, that should've been handled elsewhere787 if msg is None:788 raise SSHException("Key object may not be empty")789 # First field is always key type, in either kind of object. (make sure790 # we rewind before grabbing it - sometimes caller had to do their own791 # introspection first!)792 msg.rewind()793 type_ = msg.get_text()794 # Regular public key - nothing special to do besides the implicit795 # type check.796 if type_ in key_types:797 pass798 # OpenSSH-compatible certificate - store full copy as .public_blob799 # (so signing works correctly) and then fast-forward past the800 # nonce.801 elif type_ in cert_types:802 # This seems the cleanest way to 'clone' an already-being-read803 # message; they're *IO objects at heart and their .getvalue()804 # always returns the full value regardless of pointer position.805 self.load_certificate(Message(msg.asbytes()))806 # Read out nonce as it comes before the public numbers - our caller807 # is likely going to use the (only borrowed by us, not owned)808 # 'msg' object for loading those numbers right after this.809 # TODO: usefully interpret it & other non-public-number fields810 # (requires going back into per-type subclasses.)811 msg.get_string()812 else:813 err = "Invalid key (class: {}, data type: {}"814 raise SSHException(err.format(self.__class__.__name__, type_))815 816 def load_certificate(self, value):817 """818 Supplement the private key contents with data loaded from an OpenSSH819 public key (``.pub``) or certificate (``-cert.pub``) file, a string820 containing such a file, or a `.Message` object.821 822 The .pub contents adds no real value, since the private key823 file includes sufficient information to derive the public824 key info. For certificates, however, this can be used on825 the client side to offer authentication requests to the server826 based on certificate instead of raw public key.827 828 See:829 https://github.com/openssh/openssh-portable/blob/master/PROTOCOL.certkeys830 831 Note: very little effort is made to validate the certificate contents,832 that is for the server to decide if it is good enough to authenticate833 successfully.834 """835 if isinstance(value, Message):836 constructor = "from_message"837 elif os.path.isfile(value):838 constructor = "from_file"839 else:840 constructor = "from_string"841 blob = getattr(PublicBlob, constructor)(value)842 if not blob.key_type.startswith(self.get_name()):843 err = "PublicBlob type {} incompatible with key type {}"844 raise ValueError(err.format(blob.key_type, self.get_name()))845 self.public_blob = blob846 847 848# General construct for an OpenSSH style Public Key blob849# readable from a one-line file of the format:850# <key-name> <base64-blob> [<comment>]851# Of little value in the case of standard public keys852# {ssh-rsa, ssh-dss, ssh-ecdsa, ssh-ed25519}, but should853# provide rudimentary support for {*-cert.v01}854class PublicBlob:855 """856 OpenSSH plain public key or OpenSSH signed public key (certificate).857 858 Tries to be as dumb as possible and barely cares about specific859 per-key-type data.860 861 .. note::862 863 Most of the time you'll want to call `from_file`, `from_string` or864 `from_message` for useful instantiation, the main constructor is865 basically "I should be using ``attrs`` for this."866 """867 868 def __init__(self, type_, blob, comment=None):869 """870 Create a new public blob of given type and contents.871 872 :param str type_: Type indicator, eg ``ssh-rsa``.873 :param bytes blob: The blob bytes themselves.874 :param str comment: A comment, if one was given (e.g. file-based.)875 """876 self.key_type = type_877 self.key_blob = blob878 self.comment = comment879 880 @classmethod881 def from_file(cls, filename):882 """883 Create a public blob from a ``-cert.pub``-style file on disk.884 """885 with open(filename) as f:886 string = f.read()887 return cls.from_string(string)888 889 @classmethod890 def from_string(cls, string):891 """892 Create a public blob from a ``-cert.pub``-style string.893 """894 fields = string.split(None, 2)895 if len(fields) < 2:896 msg = "Not enough fields for public blob: {}"897 raise ValueError(msg.format(fields))898 key_type = fields[0]899 key_blob = decodebytes(b(fields[1]))900 try:901 comment = fields[2].strip()902 except IndexError:903 comment = None904 # Verify that the blob message first (string) field matches the905 # key_type906 m = Message(key_blob)907 blob_type = m.get_text()908 if blob_type != key_type:909 deets = "key type={!r}, but blob type={!r}".format(910 key_type, blob_type911 )912 raise ValueError("Invalid PublicBlob contents: {}".format(deets))913 # All good? All good.914 return cls(type_=key_type, blob=key_blob, comment=comment)915 916 @classmethod917 def from_message(cls, message):918 """919 Create a public blob from a network `.Message`.920 921 Specifically, a cert-bearing pubkey auth packet, because by definition922 OpenSSH-style certificates 'are' their own network representation."923 """924 type_ = message.get_text()925 return cls(type_=type_, blob=message.asbytes())926 927 def __str__(self):928 ret = "{} public key/certificate".format(self.key_type)929 if self.comment:930 ret += "- {}".format(self.comment)931 return ret932 933 def __eq__(self, other):934 # Just piggyback on Message/BytesIO, since both of these should be one.935 return self and other and self.key_blob == other.key_blob936 937 def __ne__(self, other):938 return not self == other939