def inflate_long(s, always_positive=False):
def deflate_long(n, add_sign_padding=True):
def format_binary(data, prefix=""):
def format_binary_line(data):
def safe_string(s):
def bit_length(n):
def tb_strings():
def generate_key_bytes(hash_alg, salt, key, nbytes):
def load_host_keys(filename):
def parse_ssh_config(file_obj):
def lookup_ssh_host_config(hostname, config):
def mod_inverse(x, m):
def get_thread_id():
def log_to_file(filename, level=DEBUG):
    def filter(self, record):
def get_logger(name):
def retry_on_signal(function):
def constant_time_bytes_eq(a, b):
    def __enter__(self):
    def __exit__(self, type, value, traceback):
def clamp_value(minimum, val, maximum):
class PFilter(object):
class ClosingContextManager(object): class SSHException(Exception):
    """
    Exception raised by failures in SSH2 protocol negotiation or logic errors.
    """

    pass class RSAKey(PKey):
    """
    Representation of an RSA key which can be used to sign and verify SSH2
    data.
    """

    HASHES = {
        "ssh-rsa": hashes.SHA1,
        "ssh-rsa-cert-v01@openssh.com": hashes.SHA1,
        "rsa-sha2-256": hashes.SHA256,
        "rsa-sha2-256-cert-v01@openssh.com": hashes.SHA256,
        "rsa-sha2-512": hashes.SHA512,
        "rsa-sha2-512-cert-v01@openssh.com": hashes.SHA512,
    }

    def __init__(
        self,
        msg=None,
        data=None,
        filename=None,
        password=None,
        key=None,
        file_obj=None,
    ):
        self.key = None
        self.public_blob = None
        if file_obj is not None:
            self._from_private_key(file_obj, password)
            return
        if filename is not None:
            self._from_private_key_file(filename, password)
            return
        if (msg is None) and (data is not None):
            msg = Message(data)
        if key is not None:
            self.key = key
        else:
            self._check_type_and_load_cert(
                msg=msg,
                # NOTE: this does NOT change when using rsa2 signatures; it's
                # purely about key loading, not exchange or verification
                key_type="ssh-rsa",
                cert_type="ssh-rsa-cert-v01@openssh.com",
            )
            self.key = rsa.RSAPublicNumbers(
                e=msg.get_mpint(), n=msg.get_mpint()
            ).public_key(default_backend())

    @property
    def size(self):
        return self.key.key_size

    @property
    def public_numbers(self):
        if isinstance(self.key, rsa.RSAPrivateKey):
            return self.key.private_numbers().public_numbers
        else:
            return self.key.public_numbers()

    def asbytes(self):
        m = Message()
        m.add_string("ssh-rsa")
        m.add_mpint(self.public_numbers.e)
        m.add_mpint(self.public_numbers.n)
        return m.asbytes()

    def __str__(self):
        # NOTE: as per inane commentary in #853, this appears to be the least
        # crummy way to get a representation that prints identical to Python
        # 2's previous behavior, on both interpreters.
        # TODO: replace with a nice clean fingerprint display or something
        if PY2:
            # Can't just return the .decode below for Py2 because stuff still
            # tries stuffing it into ASCII for whatever godforsaken reason
            return self.asbytes()
        else:
            return self.asbytes().decode("utf8", errors="ignore")

    @property
    def _fields(self):
        return (self.get_name(), self.public_numbers.e, self.public_numbers.n)

    def get_name(self):
        return "ssh-rsa"

    def get_bits(self):
        return self.size

    def can_sign(self):
        return isinstance(self.key, rsa.RSAPrivateKey)

    def sign_ssh_data(self, data, algorithm="ssh-rsa"):
        sig = self.key.sign(
            data,
            padding=padding.PKCS1v15(),
            algorithm=self.HASHES[algorithm](),
        )
        m = Message()
        m.add_string(algorithm)
        m.add_string(sig)
        return m

    def verify_ssh_sig(self, data, msg):
        sig_algorithm = msg.get_text()
        if sig_algorithm not in self.HASHES:
            return False
        key = self.key
        if isinstance(key, rsa.RSAPrivateKey):
            key = key.public_key()

        try:
            key.verify(
                msg.get_binary(),
                data,
                padding.PKCS1v15(),
                self.HASHES[sig_algorithm](),
            )
        except InvalidSignature:
            return False
        else:
            return True

    def write_private_key_file(self, filename, password=None):
        self._write_private_key_file(
            filename,
            self.key,
            serialization.PrivateFormat.TraditionalOpenSSL,
            password=password,
        )

    def write_private_key(self, file_obj, password=None):
        self._write_private_key(
            file_obj,
            self.key,
            serialization.PrivateFormat.TraditionalOpenSSL,
            password=password,
        )

    @staticmethod
    def generate(bits, progress_func=None):
        """
        Generate a new private RSA key.  This factory function can be used to
        generate a new host key or authentication key.

        :param int bits: number of bits the generated key should be.
        :param progress_func: Unused
        :return: new `.RSAKey` private key
        """
        key = rsa.generate_private_key(
            public_exponent=65537, key_size=bits, backend=default_backend()
        )
        return RSAKey(key=key)

    # ...internals...

    def _from_private_key_file(self, filename, password):
        data = self._read_private_key_file("RSA", filename, password)
        self._decode_key(data)

    def _from_private_key(self, file_obj, password):
        data = self._read_private_key("RSA", file_obj, password)
        self._decode_key(data)

    def _decode_key(self, data):
        pkformat, data = data
        if pkformat == self._PRIVATE_KEY_FORMAT_ORIGINAL:
            try:
                key = serialization.load_der_private_key(
                    data, password=None, backend=default_backend()
                )
            except (ValueError, TypeError, UnsupportedAlgorithm) as e:
                raise SSHException(str(e))
        elif pkformat == self._PRIVATE_KEY_FORMAT_OPENSSH:
            n, e, d, iqmp, p, q = self._uint32_cstruct_unpack(data, "iiiiii")
            public_numbers = rsa.RSAPublicNumbers(e=e, n=n)
            key = rsa.RSAPrivateNumbers(
                p=p,
                q=q,
                d=d,
                dmp1=d % (p - 1),
                dmq1=d % (q - 1),
                iqmp=iqmp,
                public_numbers=public_numbers,
            ).private_key(default_backend())
        else:
            self._got_bad_key_format_id(pkformat)
        assert isinstance(key, rsa.RSAPrivateKey)
        self.key = key class DSSKey(PKey):
    """
    Representation of a DSS key which can be used to sign an verify SSH2
    data.
    """

    def __init__(
        self,
        msg=None,
        data=None,
        filename=None,
        password=None,
        vals=None,
        file_obj=None,
    ):
        self.p = None
        self.q = None
        self.g = None
        self.y = None
        self.x = None
        self.public_blob = None
        if file_obj is not None:
            self._from_private_key(file_obj, password)
            return
        if filename is not None:
            self._from_private_key_file(filename, password)
            return
        if (msg is None) and (data is not None):
            msg = Message(data)
        if vals is not None:
            self.p, self.q, self.g, self.y = vals
        else:
            self._check_type_and_load_cert(
                msg=msg,
                key_type="ssh-dss",
                cert_type="ssh-dss-cert-v01@openssh.com",
            )
            self.p = msg.get_mpint()
            self.q = msg.get_mpint()
            self.g = msg.get_mpint()
            self.y = msg.get_mpint()
        self.size = util.bit_length(self.p)

    def asbytes(self):
        m = Message()
        m.add_string("ssh-dss")
        m.add_mpint(self.p)
        m.add_mpint(self.q)
        m.add_mpint(self.g)
        m.add_mpint(self.y)
        return m.asbytes()

    def __str__(self):
        return self.asbytes()

    @property
    def _fields(self):
        return (self.get_name(), self.p, self.q, self.g, self.y)

    def get_name(self):
        return "ssh-dss"

    def get_bits(self):
        return self.size

    def can_sign(self):
        return self.x is not None

    def sign_ssh_data(self, data, algorithm=None):
        key = dsa.DSAPrivateNumbers(
            x=self.x,
            public_numbers=dsa.DSAPublicNumbers(
                y=self.y,
                parameter_numbers=dsa.DSAParameterNumbers(
                    p=self.p, q=self.q, g=self.g
                ),
            ),
        ).private_key(backend=default_backend())
        sig = key.sign(data, hashes.SHA1())
        r, s = decode_dss_signature(sig)

        m = Message()
        m.add_string("ssh-dss")
        # apparently, in rare cases, r or s may be shorter than 20 bytes!
        rstr = util.deflate_long(r, 0)
        sstr = util.deflate_long(s, 0)
        if len(rstr) < 20:
            rstr = zero_byte * (20 - len(rstr)) + rstr
        if len(sstr) < 20:
            sstr = zero_byte * (20 - len(sstr)) + sstr
        m.add_string(rstr + sstr)
        return m

    def verify_ssh_sig(self, data, msg):
        if len(msg.asbytes()) == 40:
            # spies.com bug: signature has no header
            sig = msg.asbytes()
        else:
            kind = msg.get_text()
            if kind != "ssh-dss":
                return 0
            sig = msg.get_binary()

        # pull out (r, s) which are NOT encoded as mpints
        sigR = util.inflate_long(sig[:20], 1)
        sigS = util.inflate_long(sig[20:], 1)

        signature = encode_dss_signature(sigR, sigS)

        key = dsa.DSAPublicNumbers(
            y=self.y,
            parameter_numbers=dsa.DSAParameterNumbers(
                p=self.p, q=self.q, g=self.g
            ),
        ).public_key(backend=default_backend())
        try:
            key.verify(signature, data, hashes.SHA1())
        except InvalidSignature:
            return False
        else:
            return True

    def write_private_key_file(self, filename, password=None):
        key = dsa.DSAPrivateNumbers(
            x=self.x,
            public_numbers=dsa.DSAPublicNumbers(
                y=self.y,
                parameter_numbers=dsa.DSAParameterNumbers(
                    p=self.p, q=self.q, g=self.g
                ),
            ),
        ).private_key(backend=default_backend())

        self._write_private_key_file(
            filename,
            key,
            serialization.PrivateFormat.TraditionalOpenSSL,
            password=password,
        )

    def write_private_key(self, file_obj, password=None):
        key = dsa.DSAPrivateNumbers(
            x=self.x,
            public_numbers=dsa.DSAPublicNumbers(
                y=self.y,
                parameter_numbers=dsa.DSAParameterNumbers(
                    p=self.p, q=self.q, g=self.g
                ),
            ),
        ).private_key(backend=default_backend())

        self._write_private_key(
            file_obj,
            key,
            serialization.PrivateFormat.TraditionalOpenSSL,
            password=password,
        )

    @staticmethod
    def generate(bits=1024, progress_func=None):
        """
        Generate a new private DSS key.  This factory function can be used to
        generate a new host key or authentication key.

        :param int bits: number of bits the generated key should be.
        :param progress_func: Unused
        :return: new `.DSSKey` private key
        """
        numbers = dsa.generate_private_key(
            bits, backend=default_backend()
        ).private_numbers()
        key = DSSKey(
            vals=(
                numbers.public_numbers.parameter_numbers.p,
                numbers.public_numbers.parameter_numbers.q,
                numbers.public_numbers.parameter_numbers.g,
                numbers.public_numbers.y,
            )
        )
        key.x = numbers.x
        return key

    # ...internals...

    def _from_private_key_file(self, filename, password):
        data = self._read_private_key_file("DSA", filename, password)
        self._decode_key(data)

    def _from_private_key(self, file_obj, password):
        data = self._read_private_key("DSA", file_obj, password)
        self._decode_key(data)

    def _decode_key(self, data):
        pkformat, data = data
        # private key file contains:
        # DSAPrivateKey = { version = 0, p, q, g, y, x }
        if pkformat == self._PRIVATE_KEY_FORMAT_ORIGINAL:
            try:
                keylist = BER(data).decode()
            except BERException as e:
                raise SSHException("Unable to parse key file: {}".format(e))
        elif pkformat == self._PRIVATE_KEY_FORMAT_OPENSSH:
            keylist = self._uint32_cstruct_unpack(data, "iiiii")
            keylist = [0] + list(keylist)
        else:
            self._got_bad_key_format_id(pkformat)
        if type(keylist) is not list or len(keylist) < 6 or keylist[0] != 0:
            raise SSHException(
                "not a valid DSA private key file (bad ber encoding)"
            )
        self.p = keylist[1]
        self.q = keylist[2]
        self.g = keylist[3]
        self.y = keylist[4]
        self.x = keylist[5]
        self.size = util.bit_length(self.p) class ECDSAKey(PKey):
    """
    Representation of an ECDSA key which can be used to sign and verify SSH2
    data.
    """

    _ECDSA_CURVES = _ECDSACurveSet(
        [
            _ECDSACurve(ec.SECP256R1, "nistp256"),
            _ECDSACurve(ec.SECP384R1, "nistp384"),
            _ECDSACurve(ec.SECP521R1, "nistp521"),
        ]
    )

    def __init__(
        self,
        msg=None,
        data=None,
        filename=None,
        password=None,
        vals=None,
        file_obj=None,
        validate_point=True,
    ):
        self.verifying_key = None
        self.signing_key = None
        self.public_blob = None
        if file_obj is not None:
            self._from_private_key(file_obj, password)
            return
        if filename is not None:
            self._from_private_key_file(filename, password)
            return
        if (msg is None) and (data is not None):
            msg = Message(data)
        if vals is not None:
            self.signing_key, self.verifying_key = vals
            c_class = self.signing_key.curve.__class__
            self.ecdsa_curve = self._ECDSA_CURVES.get_by_curve_class(c_class)
        else:
            # Must set ecdsa_curve first; subroutines called herein may need to
            # spit out our get_name(), which relies on this.
            key_type = msg.get_text()
            # But this also means we need to hand it a real key/curve
            # identifier, so strip out any cert business. (NOTE: could push
            # that into _ECDSACurveSet.get_by_key_format_identifier(), but it
            # feels more correct to do it here?)
            suffix = "-cert-v01@openssh.com"
            if key_type.endswith(suffix):
                key_type = key_type[: -len(suffix)]
            self.ecdsa_curve = self._ECDSA_CURVES.get_by_key_format_identifier(
                key_type
            )
            key_types = self._ECDSA_CURVES.get_key_format_identifier_list()
            cert_types = [
                "{}-cert-v01@openssh.com".format(x) for x in key_types
            ]
            self._check_type_and_load_cert(
                msg=msg, key_type=key_types, cert_type=cert_types
            )
            curvename = msg.get_text()
            if curvename != self.ecdsa_curve.nist_name:
                raise SSHException(
                    "Can't handle curve of type {}".format(curvename)
                )

            pointinfo = msg.get_binary()
            try:
                key = ec.EllipticCurvePublicKey.from_encoded_point(
                    self.ecdsa_curve.curve_class(), pointinfo
                )
                self.verifying_key = key
            except ValueError:
                raise SSHException("Invalid public key")

    @classmethod
    def supported_key_format_identifiers(cls):
        return cls._ECDSA_CURVES.get_key_format_identifier_list()

    def asbytes(self):
        key = self.verifying_key
        m = Message()
        m.add_string(self.ecdsa_curve.key_format_identifier)
        m.add_string(self.ecdsa_curve.nist_name)

        numbers = key.public_numbers()

        key_size_bytes = (key.curve.key_size + 7) // 8

        x_bytes = deflate_long(numbers.x, add_sign_padding=False)
        x_bytes = b"\x00" * (key_size_bytes - len(x_bytes)) + x_bytes

        y_bytes = deflate_long(numbers.y, add_sign_padding=False)
        y_bytes = b"\x00" * (key_size_bytes - len(y_bytes)) + y_bytes

        point_str = four_byte + x_bytes + y_bytes
        m.add_string(point_str)
        return m.asbytes()

    def __str__(self):
        return self.asbytes()

    @property
    def _fields(self):
        return (
            self.get_name(),
            self.verifying_key.public_numbers().x,
            self.verifying_key.public_numbers().y,
        )

    def get_name(self):
        return self.ecdsa_curve.key_format_identifier

    def get_bits(self):
        return self.ecdsa_curve.key_length

    def can_sign(self):
        return self.signing_key is not None

    def sign_ssh_data(self, data, algorithm=None):
        ecdsa = ec.ECDSA(self.ecdsa_curve.hash_object())
        sig = self.signing_key.sign(data, ecdsa)
        r, s = decode_dss_signature(sig)

        m = Message()
        m.add_string(self.ecdsa_curve.key_format_identifier)
        m.add_string(self._sigencode(r, s))
        return m

    def verify_ssh_sig(self, data, msg):
        if msg.get_text() != self.ecdsa_curve.key_format_identifier:
            return False
        sig = msg.get_binary()
        sigR, sigS = self._sigdecode(sig)
        signature = encode_dss_signature(sigR, sigS)

        try:
            self.verifying_key.verify(
                signature, data, ec.ECDSA(self.ecdsa_curve.hash_object())
            )
        except InvalidSignature:
            return False
        else:
            return True

    def write_private_key_file(self, filename, password=None):
        self._write_private_key_file(
            filename,
            self.signing_key,
            serialization.PrivateFormat.TraditionalOpenSSL,
            password=password,
        )

    def write_private_key(self, file_obj, password=None):
        self._write_private_key(
            file_obj,
            self.signing_key,
            serialization.PrivateFormat.TraditionalOpenSSL,
            password=password,
        )

    @classmethod
    def generate(cls, curve=ec.SECP256R1(), progress_func=None, bits=None):
        """
        Generate a new private ECDSA key.  This factory function can be used to
        generate a new host key or authentication key.

        :param progress_func: Not used for this type of key.
        :returns: A new private key (`.ECDSAKey`) object
        """
        if bits is not None:
            curve = cls._ECDSA_CURVES.get_by_key_length(bits)
            if curve is None:
                raise ValueError("Unsupported key length: {:d}".format(bits))
            curve = curve.curve_class()

        private_key = ec.generate_private_key(curve, backend=default_backend())
        return ECDSAKey(vals=(private_key, private_key.public_key()))

    # ...internals...

    def _from_private_key_file(self, filename, password):
        data = self._read_private_key_file("EC", filename, password)
        self._decode_key(data)

    def _from_private_key(self, file_obj, password):
        data = self._read_private_key("EC", file_obj, password)
        self._decode_key(data)

    def _decode_key(self, data):
        pkformat, data = data
        if pkformat == self._PRIVATE_KEY_FORMAT_ORIGINAL:
            try:
                key = serialization.load_der_private_key(
                    data, password=None, backend=default_backend()
                )
            except (
                ValueError,
                AssertionError,
                TypeError,
                UnsupportedAlgorithm,
            ) as e:
                raise SSHException(str(e))
        elif pkformat == self._PRIVATE_KEY_FORMAT_OPENSSH:
            try:
                msg = Message(data)
                curve_name = msg.get_text()
                verkey = msg.get_binary()  # noqa: F841
                sigkey = msg.get_mpint()
                name = "ecdsa-sha2-" + curve_name
                curve = self._ECDSA_CURVES.get_by_key_format_identifier(name)
                if not curve:
                    raise SSHException("Invalid key curve identifier")
                key = ec.derive_private_key(
                    sigkey, curve.curve_class(), default_backend()
                )
            except Exception as e:
                # PKey._read_private_key_openssh() should check or return
                # keytype - parsing could fail for any reason due to wrong type
                raise SSHException(str(e))
        else:
            self._got_bad_key_format_id(pkformat)

        self.signing_key = key
        self.verifying_key = key.public_key()
        curve_class = key.curve.__class__
        self.ecdsa_curve = self._ECDSA_CURVES.get_by_curve_class(curve_class)

    def _sigencode(self, r, s):
        msg = Message()
        msg.add_mpint(r)
        msg.add_mpint(s)
        return msg.asbytes()

    def _sigdecode(self, sig):
        msg = Message(sig)
        r = msg.get_mpint()
        s = msg.get_mpint()
        return r, s class Ed25519Key(PKey):
    """
    Representation of an `Ed25519 <https://ed25519.cr.yp.to/>`_ key.

    .. note::
        Ed25519 key support was added to OpenSSH in version 6.5.

    .. versionadded:: 2.2
    .. versionchanged:: 2.3
        Added a ``file_obj`` parameter to match other key classes.
    """

    def __init__(
        self, msg=None, data=None, filename=None, password=None, file_obj=None
    ):
        self.public_blob = None
        verifying_key = signing_key = None
        if msg is None and data is not None:
            msg = Message(data)
        if msg is not None:
            self._check_type_and_load_cert(
                msg=msg,
                key_type="ssh-ed25519",
                cert_type="ssh-ed25519-cert-v01@openssh.com",
            )
            verifying_key = nacl.signing.VerifyKey(msg.get_binary())
        elif filename is not None:
            with open(filename, "r") as f:
                pkformat, data = self._read_private_key("OPENSSH", f)
        elif file_obj is not None:
            pkformat, data = self._read_private_key("OPENSSH", file_obj)

        if filename or file_obj:
            signing_key = self._parse_signing_key_data(data, password)

        if signing_key is None and verifying_key is None:
            raise ValueError("need a key")

        self._signing_key = signing_key
        self._verifying_key = verifying_key

    def _parse_signing_key_data(self, data, password):
        from paramiko.transport import Transport

        # We may eventually want this to be usable for other key types, as
        # OpenSSH moves to it, but for now this is just for Ed25519 keys.
        # This format is described here:
        # https://github.com/openssh/openssh-portable/blob/master/PROTOCOL.key
        # The description isn't totally complete, and I had to refer to the
        # source for a full implementation.
        message = Message(data)
        if message.get_bytes(len(OPENSSH_AUTH_MAGIC)) != OPENSSH_AUTH_MAGIC:
            raise SSHException("Invalid key")

        ciphername = message.get_text()
        kdfname = message.get_text()
        kdfoptions = message.get_binary()
        num_keys = message.get_int()

        if kdfname == "none":
            # kdfname of "none" must have an empty kdfoptions, the ciphername
            # must be "none"
            if kdfoptions or ciphername != "none":
                raise SSHException("Invalid key")
        elif kdfname == "bcrypt":
            if not password:
                raise PasswordRequiredException(
                    "Private key file is encrypted"
                )
            kdf = Message(kdfoptions)
            bcrypt_salt = kdf.get_binary()
            bcrypt_rounds = kdf.get_int()
        else:
            raise SSHException("Invalid key")

        if ciphername != "none" and ciphername not in Transport._cipher_info:
            raise SSHException("Invalid key")

        public_keys = []
        for _ in range(num_keys):
            pubkey = Message(message.get_binary())
            if pubkey.get_text() != "ssh-ed25519":
                raise SSHException("Invalid key")
            public_keys.append(pubkey.get_binary())

        private_ciphertext = message.get_binary()
        if ciphername == "none":
            private_data = private_ciphertext
        else:
            cipher = Transport._cipher_info[ciphername]
            key = bcrypt.kdf(
                password=b(password),
                salt=bcrypt_salt,
                desired_key_bytes=cipher["key-size"] + cipher["block-size"],
                rounds=bcrypt_rounds,
                # We can't control how many rounds are on disk, so no sense
                # warning about it.
                ignore_few_rounds=True,
            )
            decryptor = Cipher(
                cipher["class"](key[: cipher["key-size"]]),
                cipher["mode"](key[cipher["key-size"] :]),
                backend=default_backend(),
            ).decryptor()
            private_data = (
                decryptor.update(private_ciphertext) + decryptor.finalize()
            )

        message = Message(_unpad_openssh(private_data))
        if message.get_int() != message.get_int():
            raise SSHException("Invalid key")

        signing_keys = []
        for i in range(num_keys):
            if message.get_text() != "ssh-ed25519":
                raise SSHException("Invalid key")
            # A copy of the public key, again, ignore.
            public = message.get_binary()
            key_data = message.get_binary()
            # The second half of the key data is yet another copy of the public
            # key...
            signing_key = nacl.signing.SigningKey(key_data[:32])
            # Verify that all the public keys are the same...
            assert (
                signing_key.verify_key.encode()
                == public
                == public_keys[i]
                == key_data[32:]
            )
            signing_keys.append(signing_key)
            # Comment, ignore.
            message.get_binary()

        if len(signing_keys) != 1:
            raise SSHException("Invalid key")
        return signing_keys[0]

    def asbytes(self):
        if self.can_sign():
            v = self._signing_key.verify_key
        else:
            v = self._verifying_key
        m = Message()
        m.add_string("ssh-ed25519")
        m.add_string(v.encode())
        return m.asbytes()

    @property
    def _fields(self):
        if self.can_sign():
            v = self._signing_key.verify_key
        else:
            v = self._verifying_key
        return (self.get_name(), v)

    def get_name(self):
        return "ssh-ed25519"

    def get_bits(self):
        return 256

    def can_sign(self):
        return self._signing_key is not None

    def sign_ssh_data(self, data, algorithm=None):
        m = Message()
        m.add_string("ssh-ed25519")
        m.add_string(self._signing_key.sign(data).signature)
        return m

    def verify_ssh_sig(self, data, msg):
        if msg.get_text() != "ssh-ed25519":
            return False

        try:
            self._verifying_key.verify(data, msg.get_binary())
        except nacl.exceptions.BadSignatureError:
            return False
        else:
            return True class Message(object):
    """
    An SSH2 message is a stream of bytes that encodes some combination of
    strings, integers, bools, and infinite-precision integers (known in Python
    as longs).  This class builds or breaks down such a byte stream.

    Normally you don't need to deal with anything this low-level, but it's
    exposed for people implementing custom extensions, or features that
    paramiko doesn't support yet.
    """

    big_int = long(0xff000000)

    def __init__(self, content=None):
        """
        Create a new SSH2 message.

        :param str content:
            the byte stream to use as the message content (passed in only when
            decomposing a message).
        """
        if content is not None:
            self.packet = BytesIO(content)
        else:
            self.packet = BytesIO()

    def __str__(self):
        """
        Return the byte stream content of this message, as a string/bytes obj.
        """
        return self.asbytes()

    def __repr__(self):
        """
        Returns a string representation of this object, for debugging.
        """
        return "paramiko.Message(" + repr(self.packet.getvalue()) + ")"

    def asbytes(self):
        """
        Return the byte stream content of this Message, as bytes.
        """
        return self.packet.getvalue()

    def rewind(self):
        """
        Rewind the message to the beginning as if no items had been parsed
        out of it yet.
        """
        self.packet.seek(0)

    def get_remainder(self):
        """
        Return the bytes (as a `str`) of this message that haven't already been
        parsed and returned.
        """
        position = self.packet.tell()
        remainder = self.packet.read()
        self.packet.seek(position)
        return remainder

    def get_so_far(self):
        """
        Returns the `str` bytes of this message that have been parsed and
        returned. The string passed into a message's constructor can be
        regenerated by concatenating ``get_so_far`` and `get_remainder`.
        """
        position = self.packet.tell()
        self.rewind()
        return self.packet.read(position)

    def get_bytes(self, n):
        """
        Return the next ``n`` bytes of the message (as a `str`), without
        decomposing into an int, decoded string, etc.  Just the raw bytes are
        returned. Returns a string of ``n`` zero bytes if there weren't ``n``
        bytes remaining in the message.
        """
        b = self.packet.read(n)
        max_pad_size = 1 << 20  # Limit padding to 1 MB
        if len(b) < n < max_pad_size:
            return b + zero_byte * (n - len(b))
        return b

    def get_byte(self):
        """
        Return the next byte of the message, without decomposing it.  This
        is equivalent to `get_bytes(1) <get_bytes>`.

        :return:
            the next (`str`) byte of the message, or ``'\000'`` if there aren't
            any bytes remaining.
        """
        return self.get_bytes(1)

    def get_boolean(self):
        """
        Fetch a boolean from the stream.
        """
        b = self.get_bytes(1)
        return b != zero_byte

    def get_adaptive_int(self):
        """
        Fetch an int from the stream.

        :return: a 32-bit unsigned `int`.
        """
        byte = self.get_bytes(1)
        if byte == max_byte:
            return util.inflate_long(self.get_binary())
        byte += self.get_bytes(3)
        return struct.unpack(">I", byte)[0]

    def get_int(self):
        """
        Fetch an int from the stream.
        """
        return struct.unpack(">I", self.get_bytes(4))[0]

    def get_int64(self):
        """
        Fetch a 64-bit int from the stream.

        :return: a 64-bit unsigned integer (`long`).
        """
        return struct.unpack(">Q", self.get_bytes(8))[0]

    def get_mpint(self):
        """
        Fetch a long int (mpint) from the stream.

        :return: an arbitrary-length integer (`long`).
        """
        return util.inflate_long(self.get_binary())

    def get_string(self):
        """
        Fetch a `str` from the stream.  This could be a byte string and may
        contain unprintable characters.  (It's not unheard of for a string to
        contain another byte-stream message.)
        """
        return self.get_bytes(self.get_int())

    def get_text(self):
        """
        Fetch a Unicode string from the stream.
        """
        return u(self.get_string())

    def get_binary(self):
        """
        Fetch a string from the stream.  This could be a byte string and may
        contain unprintable characters.  (It's not unheard of for a string to
        contain another byte-stream Message.)
        """
        return self.get_bytes(self.get_int())

    def get_list(self):
        """
        Fetch a list of `strings <str>` from the stream.

        These are trivially encoded as comma-separated values in a string.
        """
        return self.get_text().split(",")

    def add_bytes(self, b):
        """
        Write bytes to the stream, without any formatting.

        :param str b: bytes to add
        """
        self.packet.write(b)
        return self

    def add_byte(self, b):
        """
        Write a single byte to the stream, without any formatting.

        :param str b: byte to add
        """
        self.packet.write(b)
        return self

    def add_boolean(self, b):
        """
        Add a boolean value to the stream.

        :param bool b: boolean value to add
        """
        if b:
            self.packet.write(one_byte)
        else:
            self.packet.write(zero_byte)
        return self

    def add_int(self, n):
        """
        Add an integer to the stream.

        :param int n: integer to add
        """
        self.packet.write(struct.pack(">I", n))
        return self

    def add_adaptive_int(self, n):
        """
        Add an integer to the stream.

        :param int n: integer to add
        """
        if n >= Message.big_int:
            self.packet.write(max_byte)
            self.add_string(util.deflate_long(n))
        else:
            self.packet.write(struct.pack(">I", n))
        return self

    def add_int64(self, n):
        """
        Add a 64-bit int to the stream.

        :param long n: long int to add
        """
        self.packet.write(struct.pack(">Q", n))
        return self

    def add_mpint(self, z):
        """
        Add a long int to the stream, encoded as an infinite-precision
        integer.  This method only works on positive numbers.

        :param long z: long int to add
        """
        self.add_string(util.deflate_long(z))
        return self

    def add_string(self, s):
        """
        Add a string to the stream.

        :param str s: string to add
        """
        s = asbytes(s)
        self.add_int(len(s))
        self.packet.write(s)
        return self

    def add_list(self, l):  # noqa: E741
        """
        Add a list of strings to the stream.  They are encoded identically to
        a single string of values separated by commas.  (Yes, really, that's
        how SSH2 does it.)

        :param l: list of strings to add
        """
        self.add_string(",".join(l))
        return self

    def _add(self, i):
        if type(i) is bool:
            return self.add_boolean(i)
        elif isinstance(i, integer_types):
            return self.add_adaptive_int(i)
        elif type(i) is list:
            return self.add_list(i)
        else:
            return self.add_string(i)

    def add(self, *seq):
        """
        Add a sequence of items to the stream.  The values are encoded based
        on their type: str, int, bool, list, or long.

        .. warning::
            Longs are encoded non-deterministically.  Don't use this method.

        :param seq: the sequence of items
        """
        for item in seq:
            self._add(item) PY2 = sys.version_info[0] < 3
        MAXSIZE = int((1 << 31) - 1)  # NOQA
        MAXSIZE = int((1 << 63) - 1)  # NOQA
    MAXSIZE = sys.maxsize  # NOQA
    def bytestring(s):  # NOQA
    def byte_mask(c, mask):
    def b(s, encoding="utf8"):  # NOQA
    def u(s, encoding="utf8"):  # NOQA
    def b2s(s):
    def is_callable(c):  # NOQA
    def get_next(c):  # NOQA
    def next(c):
        def __len__(self):
    def strftime(format, t):
    def byte_ord(c):
    def byte_chr(c):
    def byte_mask(c, mask):
    def b(s, encoding="utf8"):
    def u(s, encoding="utf8"):
    def b2s(s):
    def is_callable(c):
    def get_next(c):
    class X(object):
    class long(int): def _support(filename):
    return join(tests_dir, filename) def is_low_entropy():
    """
    Attempts to detect whether running interpreter is low-entropy.

    Classified as being in 32-bit mode under Python 2, or 32-bit mode and with
    the hash seed set to zero under Python 3.
    """
    is_32bit = struct.calcsize("P") == 32 / 8
    if PY2:
        return is_32bit
    else:
        # I don't see a way to tell internally if the hash seed was set this
        # way, but env should be plenty sufficient, this is only for testing.
        return is_32bit and os.environ.get("PYTHONHASHSEED", None) == "0"
import unittest
import os
import pytest
from binascii import hexlify
from hashlib import md5
from paramiko import (
    RSAKey,
    DSSKey,
    ECDSAKey,
    Ed25519Key,
    Message,
    util,
    SSHException,
)
from paramiko.py3compat import StringIO, byte_chr, b, bytes, PY2
from cryptography.exceptions import UnsupportedAlgorithm
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateNumbers
from mock import patch
from .util import _support, is_low_entropy
gbgeDpwfDaFJP8owByCKREocPFfi86LjCuQkyUKOfjYMN6iHIf1oEZjB8uJAatUr
FzI0ArXtUqOhwTLwTyFuUojE5own2WYsOAGByvgfyWjsGhvckYNhI4ODpNdPlxQ8
ZamaPGPsAoGARmR7CCPjodxASvRbIyzaVpZoJ/Z6x7dAumV+ysrV1BVYd0lYukmn
jO1kKBWApqpH1ve9XDQYN8zgxM4b16L21kpoWQnZtXrY3GZ4/it9kUgyB7+NwacI
BlXa8cMDL7Q/69o0d54U0X/NeX5QxuYR6OMJlrkQB7oiW/P/1mwjQgECFGI9QPSc
h9pT9XHqn+1rZ4bK+QGA
-----END DSA PRIVATE KEY-----
"""

ECDSA_PRIVATE_OUT_256 = """\
-----BEGIN EC PRIVATE KEY-----
MHcCAQEEIKB6ty3yVyKEnfF/zprx0qwC76MsMlHY4HXCnqho2eKioAoGCCqGSM49
AwEHoUQDQgAElI9mbdlaS+T9nHxY/59lFnn80EEecZDBHq4gLpccY8Mge5ZTMiMD
ADRvOqQ5R98Sxst765CAqXmRtz8vwoD96g==
-----END EC PRIVATE KEY-----
"""

ECDSA_PRIVATE_OUT_384 = """\
-----BEGIN EC PRIVATE KEY-----
MIGkAgEBBDBDdO8IXvlLJgM7+sNtPl7tI7FM5kzuEUEEPRjXIPQM7mISciwJPBt+
y43EuG8nL4mgBwYFK4EEACKhZANiAAQWxom0C1vQAGYhjdoREMVmGKBWlisDdzyk
mgyUjKpiJ9WfbIEVLsPGP8OdNjhr1y/8BZNIts+dJd6VmYw+4HzB+4F+U1Igs8K0
JEvh59VNkvWheViadDXCM2MV8Nq+DNg=
-----END EC PRIVATE KEY-----
"""

ECDSA_PRIVATE_OUT_521 = """\
-----BEGIN EC PRIVATE KEY-----
MIHcAgEBBEIAprQtAS3OF6iVUkT8IowTHWicHzShGgk86EtuEXvfQnhZFKsWm6Jo
iqAr1yEaiuI9LfB3Xs8cjuhgEEfbduYr/f6gBwYFK4EEACOhgYkDgYYABACaOaFL
ZGuxa5AW16qj6VLypFbLrEWrt9AZUloCMefxO8bNLjK/O5g0rAVasar1TnyHE9qj
4NwzANZASWjQNbc4MAG8vzqezFwLIn/kNyNTsXNfqEko9OgHZknlj2Z79dwTJcRA
L4QLcT5aND0EHZLB2fAUDXiWIb2j4rg1mwPlBMiBXA==
-----END EC PRIVATE KEY-----
"""

x1234 = b"\x01\x02\x03\x04"

TEST_KEY_BYTESTR_2 = "\x00\x00\x00\x07ssh-rsa\x00\x00\x00\x01#\x00\x00\x00\x81\x00\xd3\x8fV\xea\x07\x85\xa6k%\x8d<\x1f\xbc\x8dT\x98\xa5\x96$\xf3E#\xbe>\xbc\xd2\x93\x93\x87f\xceD\x18\xdb \x0c\xb3\xa1a\x96\xf8e#\xcc\xacS\x8a#\xefVlE\x83\x1epv\xc1o\x17M\xef\xdf\x89DUXL\xa6\x8b\xaa<\x06\x10\xd7\x93w\xec\xaf\xe2\xaf\x95\xd8\xfb\xd9\xbfw\xcb\x9f0)#y{\x10\x90\xaa\x85l\tPru\x8c\t\x19\xce\xa0\xf1\xd2\xdc\x8e/\x8b\xa8f\x9c0\xdey\x84\xd2F\xf7\xcbmm\x1f\x87"  # noqa
TEST_KEY_BYTESTR_3 = "\x00\x00\x00\x07ssh-rsa\x00\x00\x00\x01#\x00\x00\x00\x00ӏV\x07k%<\x1fT$E#>ғfD\x18 \x0cae#̬S#VlE\x1epvo\x17M߉DUXL<\x06\x10דw\u2bd5ٿw˟0)#y{\x10l\tPru\t\x19Π\u070e/f0yFmm\x1f"  # noqa


class KeyTest(unittest.TestCase):
    def setUp(self):
        pass

    def tearDown(self):
        pass

    def assert_keyfile_is_encrypted(self, keyfile):
        """
        A quick check that filename looks like an encrypted key.
        """
        with open(keyfile, "r") as fh:
            self.assertEqual(
                fh.readline()[:-1], "-----BEGIN RSA PRIVATE KEY-----"
            )
            self.assertEqual(fh.readline()[:-1], "Proc-Type: 4,ENCRYPTED")
            self.assertEqual(fh.readline()[0:10], "DEK-Info: ")

    def test_generate_key_bytes(self):
        key = util.generate_key_bytes(md5, x1234, "happy birthday", 30)
        exp = b"\x61\xE1\xF2\x72\xF4\xC1\xC4\x56\x15\x86\xBD\x32\x24\x98\xC0\xE9\x24\x67\x27\x80\xF4\x7B\xB3\x7D\xDA\x7D\x54\x01\x9E\x64"  # noqa
        self.assertEqual(exp, key)

    def test_load_rsa(self):
        key = RSAKey.from_private_key_file(_support("test_rsa.key"))
        self.assertEqual("ssh-rsa", key.get_name())
        exp_rsa = b(FINGER_RSA.split()[1].replace(":", ""))
        my_rsa = hexlify(key.get_fingerprint())
        self.assertEqual(exp_rsa, my_rsa)
        self.assertEqual(PUB_RSA.split()[1], key.get_base64())
        self.assertEqual(1024, key.get_bits())

        s = StringIO()
        key.write_private_key(s)
        self.assertEqual(RSA_PRIVATE_OUT, s.getvalue())
        s.seek(0)
        key2 = RSAKey.from_private_key(s)
        self.assertEqual(key, key2)

    def test_load_rsa_transmutes_crypto_exceptions(self):
        # TODO: nix unittest for pytest
        for exception in (TypeError("onoz"), UnsupportedAlgorithm("oops")):
            with patch(
                "paramiko.rsakey.serialization.load_der_private_key"
            ) as loader:
                loader.side_effect = exception
                with pytest.raises(SSHException, match=str(exception)):
                    RSAKey.from_private_key_file(_support("test_rsa.key"))

    def test_load_rsa_password(self):
        key = RSAKey.from_private_key_file(
            _support("test_rsa_password.key"), "television"
        )
        self.assertEqual("ssh-rsa", key.get_name())
        exp_rsa = b(FINGER_RSA.split()[1].replace(":", ""))
        my_rsa = hexlify(key.get_fingerprint())
        self.assertEqual(exp_rsa, my_rsa)
        self.assertEqual(PUB_RSA.split()[1], key.get_base64())
        self.assertEqual(1024, key.get_bits())

    def test_load_dss(self):
        key = DSSKey.from_private_key_file(_support("test_dss.key"))
        self.assertEqual("ssh-dss", key.get_name())
        exp_dss = b(FINGER_DSS.split()[1].replace(":", ""))
        my_dss = hexlify(key.get_fingerprint())
        self.assertEqual(exp_dss, my_dss)
        self.assertEqual(PUB_DSS.split()[1], key.get_base64())
        self.assertEqual(1024, key.get_bits())

        s = StringIO()
        key.write_private_key(s)
        self.assertEqual(DSS_PRIVATE_OUT, s.getvalue())
        s.seek(0)
        key2 = DSSKey.from_private_key(s)
        self.assertEqual(key, key2)

    def test_load_dss_password(self):
        key = DSSKey.from_private_key_file(

            _support("test_dss_password.key"), "television"