# ===================================================================
#
# Copyright (c) 2016, Legrandin <helderijs@gmail.com>
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
# 1. Redistributions of source code must retain the above copyright
#    notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
#    notice, this list of conditions and the following disclaimer in
#    the documentation and/or other materials provided with the
#    distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
# COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
# ===================================================================

from __future__ import annotations

from collections.abc import Sequence
from typing import TYPE_CHECKING, Callable, Optional, Union

__all__ = ["generate", "construct", "import_key", "RsaKey", "oid"]

import binascii
import math
import os
import struct
import threading

from Crypto import IncorrectPassphraseError, KdfLimitError, PassphraseRequiredError, Random
from Crypto.Math.Numbers import Integer
from Crypto.Math.Primality import COMPOSITE, generate_probable_prime, test_probable_prime
from Crypto.PublicKey import (
    _create_subject_public_key_info,
    _expand_subject_public_key_info,
    _extract_subject_public_key_info,
)
from Crypto.Util._bytes import tobytes
from Crypto.Util.asn1 import DerNull, DerSequence
from Crypto.Util.number import bytes_to_long

# A random blinding factor r is used for this many private operations:
# in between, it is squared (like in OpenSSL), which is much faster than
# a new r^e and r^-1.
_BLINDING_USES = 32

# Increased in a child process after a fork(), so that it does not reuse
# the blinding factors of the parent
_fork_generation = [0]


def _after_fork_in_child() -> None:
    _fork_generation[0] += 1


if hasattr(os, "register_at_fork"):
    os.register_at_fork(after_in_child=_after_fork_in_child)

# The largest RSA modulus accepted, in bits. The cost of the operations
# grows quickly with the size, and a public key may come from anyone.
MAX_MODULUS_BITS = 16384

if TYPE_CHECKING:
    from Crypto.IO._PBES import ProtParams

RNG = Callable[[int], bytes]
Int = Union[int, Integer]


class RsaKey:
    r"""Class defining an RSA key, private or public.
    Do not instantiate directly.
    Use :func:`generate`, :func:`construct` or :func:`import_key` instead.

    :ivar n: RSA modulus
    :vartype n: integer

    :ivar e: RSA public exponent
    :vartype e: integer

    :ivar d: RSA private exponent
    :vartype d: integer

    :ivar p: First factor of the RSA modulus
    :vartype p: integer

    :ivar q: Second factor of the RSA modulus
    :vartype q: integer

    :ivar invp: Chinese remainder component (:math:`p^{-1} \text{mod } q`)
    :vartype invp: integer

    :ivar invq: Chinese remainder component (:math:`q^{-1} \text{mod } p`)
    :vartype invq: integer

    :ivar u: Same as ``invp``
    :vartype u: integer
    """

    # The key components are set dynamically in the constructor
    _n: Integer
    _e: Integer
    _d: Integer
    _p: Integer
    _q: Integer
    _u: Integer
    _dp: Integer
    _dq: Integer
    _invq: Optional[Integer]
    # CRT components for decryption, ordered so that _crt_p < _crt_q
    _crt_p: Integer
    _crt_q: Integer
    _crt_dp: Integer
    _crt_dq: Integer
    _crt_u: Integer
    # (r^e mod n, r^-1 mod n, uses left, fork generation), see _blinding_pair()
    _blinding: Optional[tuple[Integer, Integer, int, int]]
    _blinding_lock: threading.Lock

    def __init__(self, **kwargs: Int) -> None:
        """Build an RSA key.

        :Keywords:
          n : integer
            The modulus.
          e : integer
            The public exponent.
          d : integer
            The private exponent. Only required for private keys.
          p : integer
            The first factor of the modulus. Only required for private keys.
          q : integer
            The second factor of the modulus. Only required for private keys.
          u : integer
            The CRT coefficient (inverse of p modulo q). Only required for
            private keys.
        """

        input_set = set(kwargs.keys())
        public_set = {"n", "e"}
        private_set = public_set | {"p", "q", "d", "u"}
        if input_set not in (private_set, public_set):
            raise ValueError("Some RSA components are missing")
        for component, value in kwargs.items():
            setattr(self, "_" + component, value)
        # The modulus and the public exponent are public: the bounds of
        # random_range() and the powers with e can use their values
        self._n = Integer(self._n, public=True)
        self._e = Integer(self._e, public=True)
        if input_set == private_set:
            self._dp = self._d % (self._p - 1)  # = (e⁻¹) mod (p-1)
            self._dq = self._d % (self._q - 1)  # = (e⁻¹) mod (q-1)
            self._invq = None  # will be computed on demand
            self._init_crt()
            self._blinding = None
            self._blinding_lock = threading.Lock()

    def _init_crt(self) -> None:
        """Prepare the CRT components used for decryption.

        They are ordered so that p < q, as the CRT step needs m1 < q, with
        m1 < p. This comparison only reveals which prime is the larger one.
        The private exponents get the same size as the primes, so that
        their actual length does not leak.
        """
        p, q = Integer(self._p), Integer(self._q)
        if not q < p:
            dp, dq, u = self._dp, self._dq, Integer(self._u)
        else:
            p, q = q, p
            dp, dq, u = self._dq, self._dp, p.inverse(q)
        self._crt_p, self._crt_q, self._crt_u = p, q, u
        self._crt_dp = Integer(dp)._resize(p.size_in_bits())
        self._crt_dq = Integer(dq)._resize(q.size_in_bits())

    @property
    def n(self) -> int:
        return int(self._n)

    @property
    def e(self) -> int:
        return int(self._e)

    @property
    def d(self) -> int:
        if not self.has_private():
            raise AttributeError("No private exponent available for public keys")
        return int(self._d)

    @property
    def p(self) -> int:
        if not self.has_private():
            raise AttributeError("No CRT component 'p' available for public keys")
        return int(self._p)

    @property
    def q(self) -> int:
        if not self.has_private():
            raise AttributeError("No CRT component 'q' available for public keys")
        return int(self._q)

    @property
    def dp(self):
        if not self.has_private():
            raise AttributeError("No CRT component 'dp' available for public keys")
        return int(self._dp)

    @property
    def dq(self):
        if not self.has_private():
            raise AttributeError("No CRT component 'dq' available for public keys")
        return int(self._dq)

    @property
    def invq(self) -> int:
        if not self.has_private():
            raise AttributeError("No CRT component 'invq' available for public keys")
        if self._invq is None:
            self._invq = self._q.inverse(self._p)
        return int(self._invq)

    @property
    def invp(self) -> int:
        return self.u

    @property
    def u(self) -> int:
        if not self.has_private():
            raise AttributeError("No CRT component 'u' available for public keys")
        return int(self._u)

    def size_in_bits(self) -> int:
        """Size of the RSA modulus in bits"""
        return self._n.size_in_bits()

    def size_in_bytes(self) -> int:
        """The minimal amount of bytes that can hold the RSA modulus"""
        return (self._n.size_in_bits() - 1) // 8 + 1

    def _encrypt(self, plaintext):
        if not 0 <= plaintext < self._n:
            raise ValueError("Plaintext too large")
        return int(pow(Integer(plaintext), self._e, self._n))

    def _decrypt_to_bytes(self, ciphertext):
        if not 0 <= ciphertext < self._n:
            raise ValueError("Ciphertext too large")
        if not self.has_private():
            raise TypeError("This is not a private key")

        # Blinded RSA decryption (to prevent timing attacks):
        # Step 1: Get a secret blinding factor r, with 0 < r < n
        r_e, r_inv = self._blinding_pair()
        # Step 2: Compute c' = c * r**e mod n
        cp = Integer(ciphertext)._mul_mod(r_e, self._n)
        # Step 3: Compute m' = c'**d mod n       (normal RSA decryption)
        m1 = pow(cp, self._crt_dp, self._crt_p)
        m2 = pow(cp, self._crt_dq, self._crt_q)
        # (m2 - m1) mod q, with m1 < p < q
        h = m2._sub_mod(m1, self._crt_q)._mul_mod(self._crt_u, self._crt_q)
        mp = h * self._crt_p + m1
        # Step 4: Compute m = m' * (r**(-1)) mod n
        # then encode into a big endian byte string
        result = Integer._mult_modulo_bytes(r_inv, mp, self._n)
        return result

    def _blinding_pair(self) -> tuple[Integer, Integer]:
        """A pair (r^e mod n, r^-1 mod n), for a different secret r at each call.

        Every _BLINDING_USES calls (and after a fork), r is a new random
        number in [1, n). Otherwise, it is the square of the previous r.
        """
        n = self._n
        with self._blinding_lock:
            state = self._blinding
            if state is None or state[2] == 0 or state[3] != _fork_generation[0]:
                r = Integer.random_range(min_inclusive=1, max_exclusive=n)
                r_e, r_inv, uses = pow(r, self._e, n), r.inverse(n), _BLINDING_USES
            else:
                r_e, r_inv, uses = state[0]._mul_mod(state[0], n), state[1]._mul_mod(state[1], n), state[2]
            self._blinding = (r_e, r_inv, uses - 1, _fork_generation[0])
        return r_e, r_inv

    def _decrypt(self, ciphertext):
        """Legacy private method"""

        return bytes_to_long(self._decrypt_to_bytes(ciphertext))

    def has_private(self) -> bool:
        """Whether this is an RSA private key"""

        return hasattr(self, "_d")

    def can_encrypt(self) -> bool:  # legacy
        return True

    def can_sign(self) -> bool:  # legacy
        return True

    def public_key(self) -> RsaKey:
        """A matching RSA public key.

        Returns:
            a new :class:`RsaKey` object
        """
        return RsaKey(n=self._n, e=self._e)

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, RsaKey):
            return False
        if self.has_private() != other.has_private():
            return False
        if self.n != other.n or self.e != other.e:
            return False
        if not self.has_private():
            return True
        return self.d == other.d

    def __getstate__(self) -> None:
        # RSA key is not pickable
        from pickle import PicklingError

        raise PicklingError

    def __repr__(self) -> str:
        if self.has_private():
            extra = ", d=%d, p=%d, q=%d, u=%d" % (int(self._d), int(self._p), int(self._q), int(self._u))
        else:
            extra = ""
        return "RsaKey(n=%d, e=%d%s)" % (int(self._n), int(self._e), extra)

    def __str__(self) -> str:
        if self.has_private():
            key_type = "Private"
        else:
            key_type = "Public"
        return "%s RSA key at 0x%X" % (key_type, id(self))

    def export_key(
        self,
        format: str = "PEM",
        passphrase: Optional[Union[bytes, str]] = None,
        pkcs: Optional[int] = None,
        protection: Optional[str] = None,
        randfunc: Optional[RNG] = None,
        prot_params: Optional[ProtParams] = None,
        *,
        threads: int = 1,
    ) -> bytes:
        """Export this RSA key.

        Keyword Args:
          format (string):
            The desired output format:

            - ``'PEM'``. (default) Text output, according to `RFC1421`_/`RFC1423`_.
            - ``'DER'``. Binary output.
            - ``'OpenSSH'``. Text output, according to the OpenSSH specification.
              Only suitable for public keys (not private keys).

            Note that PEM contains a DER structure.

          passphrase (bytes or string):
            (*Private keys only*) The passphrase to protect the
            private key.

          pkcs (integer):
            (*Private keys only*) The standard to use for
            serializing the key: PKCS#1 or PKCS#8.

            With ``pkcs=1``, the private key is encoded with a
            simple `PKCS#1`_ structure (``RSAPrivateKey``). The key cannot be
            securely encrypted: with a ``passphrase`` (and ``format='PEM'``),
            the obsolete PEM encryption is used, which derives the key with
            a single round of MD5 and is therefore easy to brute force.
            Use it only with a high-entropy passphrase (like 16 random bytes),
            for compatibility with old software.

            With ``pkcs=8``, the private key is encoded with a `PKCS#8`_ structure
            (``PrivateKeyInfo``). PKCS#8 offers the best ways to securely
            encrypt the key.

            If ``None`` (default), PKCS#1 is used for a key in the clear,
            and PKCS#8 when there is a ``passphrase`` (that is, the key
            is always securely encrypted).

            .. note::
                This parameter is ignored for a public key.
                For DER and PEM, the output is always an
                ASN.1 DER ``SubjectPublicKeyInfo`` structure.

          protection (string):
            (*For private keys only*)
            The encryption scheme to use for protecting the private key
            using the passphrase.

            It is only allowed with PKCS#8 (not with ``pkcs=1``).
            For all possible protection schemes,
            refer to :ref:`the encryption parameters of PKCS#8<enc_params>`.
            The default is ``'PBKDF2WithHMAC-SHA256AndAES256-CBC'``.

          prot_params (dict):
            (*For private keys only*)

            The parameters to use to derive the encryption key
            from the passphrase. It is only allowed with PKCS#8.
            For all possible values,
            refer to :ref:`the encryption parameters of PKCS#8<enc_params>`.
            The defaults meet the OWASP recommendations (with the default
            ``protection``): there is no need to set them.

          randfunc (callable):
            A function that provides random bytes. Only used for PEM encoding.
            The default is :func:`Crypto.Random.get_random_bytes`.

          threads (integer):
            (*For private keys only*)
            The maximum number of threads used to derive the encryption key
            with scrypt (default: 1, no extra threads; 0 for all CPU cores).
            Only useful with ``pkcs=8``, a scrypt ``protection`` and a
            ``parallelization`` larger than 1 in ``prot_params``.
            The output does not depend on the number of threads.

        Returns:
          bytes: the encoded key

        Raises:
          ValueError:when the format is unknown, when you try to encrypt a private
            key with *DER* format and PKCS#1, when you pass ``protection``
            or ``prot_params`` with PKCS#1 or without a passphrase,
            or when the passphrase is empty.

        .. warning::
            If you don't provide a pass phrase, the private key will be
            exported in the clear!

        .. _RFC1421:    https://www.rfc-editor.org/rfc/rfc1421
        .. _RFC1423:    https://www.rfc-editor.org/rfc/rfc1423
        .. _`PKCS#1`:   https://www.rfc-editor.org/rfc/rfc8017
        .. _`PKCS#8`:   https://www.rfc-editor.org/rfc/rfc5958
        """

        from Crypto.IO import PKCS8

        PKCS8._check_kdf_params(threads=threads)
        if passphrase is not None:
            passphrase = tobytes(passphrase)

        if randfunc is None:
            randfunc = Random.get_random_bytes

        if format == "OpenSSH":
            e_bytes, n_bytes = (x.to_bytes() for x in (self._e, self._n))
            if e_bytes[0] & 0x80:
                e_bytes = b"\x00" + e_bytes
            if n_bytes[0] & 0x80:
                n_bytes = b"\x00" + n_bytes
            keyparts = [b"ssh-rsa", e_bytes, n_bytes]
            keystring = b"".join([struct.pack(">I", len(kp)) + kp for kp in keyparts])
            return b"ssh-rsa " + binascii.b2a_base64(keystring)[:-1]

        # DER format is always used, even in case of PEM, which simply
        # encodes it into BASE64.
        if self.has_private():
            if passphrase is None:
                if protection or prot_params:
                    raise ValueError("'protection' and 'prot_params' require a passphrase")
            elif not passphrase:
                raise ValueError("Empty passphrase")
            binary_key = DerSequence(
                [
                    0,
                    self.n,
                    self.e,
                    self.d,
                    self.p,
                    self.q,
                    self.d % (self.p - 1),
                    self.d % (self.q - 1),
                    int(Integer(self.q).inverse(self.p)),
                ]
            ).encode()
            if pkcs is None:
                pkcs = 1 if passphrase is None else 8
            if pkcs == 1:
                key_type = "RSA PRIVATE KEY"
                if protection or prot_params:
                    raise ValueError("'protection' and 'prot_params' are only supported for PKCS#8")
                if format == "DER" and passphrase:
                    raise ValueError("PKCS#1 private key cannot be encrypted")
            else:  # PKCS#8
                if passphrase is None:
                    key_type = "PRIVATE KEY"
                    binary_key = PKCS8.wrap(binary_key, oid, None, key_params=DerNull())
                else:
                    key_type = "ENCRYPTED PRIVATE KEY"
                    binary_key = PKCS8.wrap(
                        binary_key,
                        oid,
                        passphrase,
                        protection,
                        prot_params=prot_params,
                        key_params=DerNull(),
                        threads=threads,
                    )
                    passphrase = None
        else:
            key_type = "PUBLIC KEY"
            binary_key = _create_subject_public_key_info(oid, DerSequence([self.n, self.e]), DerNull())

        if format == "DER":
            return binary_key
        if format == "PEM":
            from Crypto.IO import PEM

            pem_str = PEM.encode(binary_key, key_type, passphrase, randfunc)
            return tobytes(pem_str)

        raise ValueError("Unknown key format '%s'. Cannot export the RSA key." % format)

    # Backward compatibility
    def exportKey(self, *args, **kwargs):
        """:meta private:"""
        return self.export_key(*args, **kwargs)

    def publickey(self):
        """:meta private:"""
        return self.public_key()

    # Methods defined in PyCrypto that we don't support anymore
    def sign(self, M, K):
        """:meta private:"""
        raise NotImplementedError("Use module Crypto.Signature.pkcs1_15 instead")

    def verify(self, M, signature):
        """:meta private:"""
        raise NotImplementedError("Use module Crypto.Signature.pkcs1_15 instead")

    def encrypt(self, plaintext, K):
        """:meta private:"""
        raise NotImplementedError("Use module Crypto.Cipher.oaep instead")

    def decrypt(self, ciphertext):
        """:meta private:"""
        raise NotImplementedError("Use module Crypto.Cipher.oaep instead")

    def blind(self, M, B):
        """:meta private:"""
        raise NotImplementedError

    def unblind(self, M, B):
        """:meta private:"""
        raise NotImplementedError

    def size(self):
        """:meta private:"""
        raise NotImplementedError


def generate(bits: int, randfunc: Optional[RNG] = None, e: Int = 65537) -> RsaKey:
    """Create a new RSA key pair.

    The algorithm closely follows NIST `FIPS 186-4`_ in its
    sections B.3.1 and B.3.3. The modulus is the product of
    two non-strong probable primes.
    Each prime passes a suitable number of Miller-Rabin tests
    with random bases and a single Lucas test.

    Args:
      bits (integer):
        Key length, or size (in bits) of the RSA modulus.
        It must be at least 1024, but **2048 is recommended.**
        The FIPS standard only defines 1024, 2048 and 3072.
        It cannot be larger than 16384.
    Keyword Args:
      randfunc (callable):
        Function that returns random bytes.
        The default is :func:`Crypto.Random.get_random_bytes`.
      e (integer):
        Public RSA exponent. It must be an odd positive integer.
        It is typically a small number with very few ones in its
        binary representation.
        The FIPS standard requires the public exponent to be
        at least 65537 (the default).

    Returns: an RSA key object (:class:`RsaKey`, with private key).

    .. _FIPS 186-4: http://nvlpubs.nist.gov/nistpubs/FIPS/NIST.FIPS.186-4.pdf
    """

    if bits < 1024:
        raise ValueError("RSA modulus length must be >= 1024")
    if bits > MAX_MODULUS_BITS:
        raise ValueError("RSA modulus length must be <= %d" % MAX_MODULUS_BITS)
    if e % 2 == 0 or e < 3:
        raise ValueError("RSA public exponent must be a positive, odd integer larger than 2.")

    if randfunc is None:
        randfunc = Random.get_random_bytes

    d = n = Integer(1)
    e = Integer(e)

    while n.size_in_bits() != bits or d < (1 << (bits // 2)):
        # Generate the prime factors of n: p and q.
        # By construciton, their product is always
        # 2^{bits-1} < p*q < 2^bits.
        size_q = bits // 2
        size_p = bits - size_q

        min_p = min_q = (Integer(1) << (2 * size_q - 1)).sqrt()
        if size_q != size_p:
            min_p = (Integer(1) << (2 * size_p - 1)).sqrt()

        def filter_p(candidate):
            return candidate > min_p and (candidate - 1).gcd(e) == 1  # noqa: B023 (used in this iteration)

        p = generate_probable_prime(exact_bits=size_p, randfunc=randfunc, prime_filter=filter_p)

        min_distance = Integer(1) << (bits // 2 - 100)

        def filter_q(candidate):
            return (
                candidate > min_q  # noqa: B023 (used in this iteration)
                and (candidate - 1).gcd(e) == 1
                and (candidate - p if candidate > p else p - candidate) > min_distance  # noqa: B023
            )

        q = generate_probable_prime(exact_bits=size_q, randfunc=randfunc, prime_filter=filter_q)

        n = p * q
        lcm = (p - 1).lcm(q - 1)
        d = e.inverse(lcm)

    if p > q:
        p, q = q, p

    u = p.inverse(q)

    return RsaKey(n=n, e=e, d=d, p=p, q=q, u=u)


def construct(
    rsa_components: Union[
        tuple[Int, Int],  # n, e
        tuple[Int, Int, Int],  # n, e, d
        tuple[Int, Int, Int, Int, Int],  # n, e, d, p, q
        tuple[Int, Int, Int, Int, Int, Int],  # n, e, d, p, q, crt_q
        Sequence[Int],
    ],
    consistency_check: bool = True,
) -> RsaKey:
    r"""Construct an RSA key from a tuple of valid RSA components.

    The modulus **n** must be the product of two primes,
    and it cannot be larger than 16384 bits.
    The public exponent **e** must be odd and larger than 1.

    In case of a private key, the following equations must apply:

    .. math::

        \begin{align}
        p*q &= n \\
        e*d &\equiv 1 ( \text{mod lcm} [p-1, q-1]) \\
        p*u &\equiv 1 ( \text{mod } q)
        \end{align}

    Args:
        rsa_components (tuple):
            A tuple of integers, with at least 2 and no
            more than 6 items. The items come in the following order:

            1. RSA modulus *n*.
            2. Public exponent *e*.
            3. Private exponent *d*.
               Only required if the key is private.
            4. First factor of *n* (*p*).
               Optional, but the other factor *q* must also be present.
            5. Second factor of *n* (*q*). Optional.
            6. CRT coefficient *u*, that is :math:`p^{-1} \text{mod }q`. Optional.

    Keyword Args:
        consistency_check (boolean):
            If ``True``, the library will verify that the provided components
            fulfil the main RSA properties.

    Raises:
        ValueError: when the key being imported fails the most basic RSA validity checks.

    Returns: An RSA key object (:class:`RsaKey`).
    """

    input_comps = {
        comp: Integer(value) for (comp, value) in zip(("n", "e", "d", "p", "q", "u"), rsa_components)
    }

    n = input_comps["n"]
    e = input_comps["e"]
    # Also without the consistency check: n is public, and it bounds the cost of everything else
    if n.size_in_bits() > MAX_MODULUS_BITS:
        raise ValueError("RSA modulus is larger than %d bits" % MAX_MODULUS_BITS)
    if "d" not in input_comps:
        key = RsaKey(n=n, e=e)
    else:
        d = input_comps["d"]
        if "q" in input_comps:
            p = input_comps["p"]
            q = input_comps["q"]
        else:
            # Compute factors p and q from the private exponent d.
            # We assume that n has no more than two factors.
            # See 8.2.2(i) in Handbook of Applied Cryptography.
            ktot = d * e - 1
            # The quantity d*e-1 is a multiple of phi(n), even,
            # and can be represented as t*2^s.
            t = ktot
            while t % 2 == 0:
                t //= 2
            # Cycle through all multiplicative inverses in Zn.
            # The algorithm is non-deterministic, but there is a 50% chance
            # any candidate a leads to successful factoring.
            # See "Digitalized Signatures and Public Key Functions as Intractable
            # as Factorization", M. Rabin, 1979
            spotted = False
            a = Integer(2)
            while not spotted and a < 100:
                k = Integer(t)
                # Cycle through all values a^{t*2^i}=a^k
                while k < ktot:
                    cand = pow(a, k, n)
                    # Check if a^k is a non-trivial root of unity (mod n)
                    if cand != 1 and cand != (n - 1) and pow(cand, 2, n) == 1:
                        # We have found a number such that (cand-1)(cand+1)=0 (mod n).
                        # Either of the terms divides n.
                        p = Integer(n).gcd(cand + 1)
                        spotted = True
                        break
                    k *= 2
                # This value was not any good... let's try another!
                a += 2
            if not spotted:
                raise ValueError("Unable to compute factors p and q from exponent d.")
            # Found !
            assert (n % p) == 0
            q = n // p

        if "u" in input_comps:
            u = input_comps["u"]
        else:
            u = p.inverse(q)

        # Build key object
        key = RsaKey(n=n, e=e, d=d, p=p, q=q, u=u)

    # Verify consistency of the key
    if consistency_check:
        # Modulus and public exponent must be coprime
        if e <= 1 or e >= n:
            raise ValueError("Invalid RSA public exponent")
        if math.gcd(int(n), int(e)) != 1:
            raise ValueError("RSA public exponent is not coprime to modulus")

        # For RSA, modulus must be odd
        if not n & 1:
            raise ValueError("RSA modulus is not odd")

        if key.has_private():
            # Modulus and private exponent must be coprime
            if d <= 1 or d >= n:
                raise ValueError("Invalid RSA private exponent")
            if Integer(n).gcd(d) != 1:
                raise ValueError("RSA private exponent is not coprime to modulus")
            # Modulus must be product of 2 primes
            if p * q != n:
                raise ValueError("RSA factors do not match modulus")
            if test_probable_prime(p) == COMPOSITE:
                raise ValueError("RSA factor p is composite")
            if test_probable_prime(q) == COMPOSITE:
                raise ValueError("RSA factor q is composite")
            # See Carmichael theorem
            phi = (p - 1) * (q - 1)
            lcm = phi // (p - 1).gcd(q - 1)
            if (e * d % lcm) != 1:
                raise ValueError("Invalid RSA condition")
            if hasattr(key, "u"):
                # CRT coefficient
                if u <= 1 or u >= q:
                    raise ValueError("Invalid RSA component u")
                if (p * u % q) != 1:
                    raise ValueError("Invalid RSA component u with p")

    return key


def _import_pkcs1_private(encoded, *kwargs):
    # RSAPrivateKey ::= SEQUENCE {
    #           version Version,
    #           modulus INTEGER, -- n
    #           publicExponent INTEGER, -- e
    #           privateExponent INTEGER, -- d
    #           prime1 INTEGER, -- p
    #           prime2 INTEGER, -- q
    #           exponent1 INTEGER, -- d mod (p-1)
    #           exponent2 INTEGER, -- d mod (q-1)
    #           coefficient INTEGER -- (inverse of q) mod p
    # }
    #
    # Version ::= INTEGER
    der = DerSequence().decode(encoded, nr_elements=9, only_ints_expected=True)
    if der[0] != 0:
        raise ValueError("No PKCS#1 encoding of an RSA private key")
    return construct(der[1:6] + [Integer(der[4]).inverse(der[5])])


def _import_pkcs1_public(encoded, *kwargs):
    # RSAPublicKey ::= SEQUENCE {
    #           modulus INTEGER, -- n
    #           publicExponent INTEGER -- e
    # }
    der = DerSequence().decode(encoded, nr_elements=2, only_ints_expected=True)
    return construct(der)


def _import_subjectPublicKeyInfo(encoded, *kwargs):
    oids = (oid, "1.2.840.113549.1.1.10")

    algoid, encoded_key, params = _expand_subject_public_key_info(encoded)
    if algoid not in oids or params is not None:
        raise ValueError("No RSA subjectPublicKeyInfo")
    return _import_pkcs1_public(encoded_key)


def _import_x509_cert(encoded, *kwargs):
    sp_info = _extract_subject_public_key_info(encoded)
    return _import_subjectPublicKeyInfo(sp_info)


def _import_pkcs8(encoded, passphrase, max_iteration_count=None, threads=1, max_memory=None):
    from Crypto.IO import PKCS8

    oids = (oid, "1.2.840.113549.1.1.10")

    k = PKCS8.unwrap(encoded, passphrase, max_iteration_count, threads=threads, max_memory=max_memory)
    if k[0] not in oids:
        raise ValueError("No PKCS#8 encoded RSA key")
    # The private key inside PKCS#8 is an RSAPrivateKey (RFC 8017),
    # and never another (possibly encrypted) container
    return _import_pkcs1_private(k[1])


def _import_keyDER(extern_key, passphrase, max_iteration_count=None, threads=1, max_memory=None):
    """Import an RSA key (public or private half), encoded in DER form."""

    decodings = (
        _import_pkcs1_private,
        _import_pkcs1_public,
        _import_subjectPublicKeyInfo,
        _import_x509_cert,
    )

    for decoding in decodings:
        try:
            return decoding(extern_key, passphrase)
        except ValueError:
            pass

    try:
        return _import_pkcs8(extern_key, passphrase, max_iteration_count, threads, max_memory)
    except (KdfLimitError, PassphraseRequiredError, IncorrectPassphraseError):
        raise
    except ValueError:
        pass

    raise ValueError("RSA key format is not supported")


def _import_openssh_private_rsa(data, password, max_iteration_count=None):
    from ._openssh import check_padding, import_openssh_private_generic, read_bytes, read_string

    ssh_name, decrypted = import_openssh_private_generic(data, password, max_iteration_count)

    if ssh_name != "ssh-rsa":
        raise ValueError("This SSH key is not RSA")

    n, decrypted = read_bytes(decrypted)
    e, decrypted = read_bytes(decrypted)
    d, decrypted = read_bytes(decrypted)
    iqmp, decrypted = read_bytes(decrypted)
    p, decrypted = read_bytes(decrypted)
    q, decrypted = read_bytes(decrypted)

    _, padded = read_string(decrypted)  # Comment
    check_padding(padded)

    build = [Integer.from_bytes(x) for x in (n, e, d, q, p, iqmp)]
    return construct(build)


def import_key(
    extern_key: Union[str, bytes],
    passphrase: Optional[Union[str, bytes]] = None,
    *,
    max_iteration_count: Optional[int] = None,
    threads: int = 1,
    max_memory: Optional[int] = None,
) -> RsaKey:
    """Import an RSA key (public or private).

    Args:
      extern_key (string or byte string):
        The RSA key to import.

        The following formats are supported for an RSA **public key**:

        - X.509 certificate (binary or PEM format)
        - X.509 ``subjectPublicKeyInfo`` DER SEQUENCE (binary or PEM
          encoding)
        - `PKCS#1`_ ``RSAPublicKey`` DER SEQUENCE (binary or PEM encoding)
        - An OpenSSH line (e.g. the content of ``~/.ssh/id_ecdsa``, ASCII)

        The following formats are supported for an RSA **private key**:

        - PKCS#1 ``RSAPrivateKey`` DER SEQUENCE (binary or PEM encoding)
        - `PKCS#8`_ ``PrivateKeyInfo`` or ``EncryptedPrivateKeyInfo``
          DER SEQUENCE (binary or PEM encoding)
        - OpenSSH (text format, introduced in `OpenSSH 6.5`_)

        For details about the PEM encoding, see `RFC1421`_/`RFC1423`_.

      passphrase (string or byte string):
        For private keys only, the pass phrase that encrypts the key.

      max_iteration_count (integer):
        The maximum value allowed for the PBKDF2 iteration count or
        the scrypt cost *N·r·p/8* (that is, *N* for the usual *r=8* and *p=1*)
        of a `PKCS#8`_ encrypted key, or for the number of rounds of
        bcrypt_pbkdf of an encrypted OpenSSH key.
        A malicious key could require an extremely high number of
        iterations, tying up the CPU for hours.
        The default value is None: a limit for each KDF, which keeps
        the key derivation to about one minute at most
        (see :ref:`the limits for decryption<pkcs8_limits>`).
        Set it to ``0`` to disable the check.

      threads (integer):
        The maximum number of threads used to derive the decryption key
        with scrypt (default: 1, no extra threads; 0 for all CPU cores).
        Only a `PKCS#8`_ key protected with scrypt and a parallelization
        parameter larger than 1 can use them.

      max_memory (integer):
        The maximum memory (in bytes) for the key derivation of a `PKCS#8`_
        encrypted key, as a limit on the **total** of all threads together
        (not on each thread).
        Only scrypt uses a significant amount: 128 · (*N* + 2) · *r* bytes for
        **each** thread, plus 256 · *r* · *p* bytes for all the flows together.
        If the requested threads do not fit, fewer are used (down to one);
        if a single thread does not fit, the key is rejected.
        The default value is None (a limit of 256 MiB).
        Set it to ``0`` to disable the check.

    Returns: An RSA key object (:class:`RsaKey`).

    Raises:
      PassphraseRequiredError:
        (a ``ValueError``) when the private key is encrypted,
        but no passphrase was provided.
      IncorrectPassphraseError:
        (a ``ValueError``) when the private key cannot be decrypted
        with the passphrase (most likely, the passphrase is wrong).
      ValueError/IndexError/TypeError:
        When the given key cannot be parsed.

    .. _RFC1421: https://www.rfc-editor.org/rfc/rfc1421
    .. _RFC1423: https://www.rfc-editor.org/rfc/rfc1423
    .. _`PKCS#1`: https://www.rfc-editor.org/rfc/rfc8017
    .. _`PKCS#8`: https://www.rfc-editor.org/rfc/rfc5958
    .. _`OpenSSH 6.5`: https://flak.tedunangst.com/post/new-openssh-key-format-and-bcrypt-pbkdf
    """

    from Crypto.IO import PEM, PKCS8

    PKCS8._check_kdf_params(max_iteration_count, threads, max_memory)
    extern_key = tobytes(extern_key)
    if passphrase is not None:
        passphrase = tobytes(passphrase)

    if extern_key.startswith(b"-----BEGIN OPENSSH PRIVATE KEY"):
        text_encoded = extern_key.decode("latin-1")
        openssh_encoded, _marker, enc_flag = PEM.decode(text_encoded, passphrase)
        result = _import_openssh_private_rsa(openssh_encoded, passphrase, max_iteration_count)
        return result

    if extern_key.startswith(b"-----"):
        # This is probably a PEM encoded key.
        (der, _marker, enc_flag) = PEM.decode(extern_key.decode("latin-1"), passphrase)
        if enc_flag:
            passphrase = None
        return _import_keyDER(der, passphrase, max_iteration_count, threads, max_memory)

    if extern_key.startswith(b"ssh-rsa "):
        # This is probably an OpenSSH key
        keystring = binascii.a2b_base64(extern_key.split(b" ")[1])
        keyparts = []
        while len(keystring) > 4:
            length = struct.unpack(">I", keystring[:4])[0]
            keyparts.append(keystring[4 : 4 + length])
            keystring = keystring[4 + length :]
        e = Integer.from_bytes(keyparts[1])
        n = Integer.from_bytes(keyparts[2])
        return construct([n, e])

    if len(extern_key) > 0 and extern_key[0] == 0x30:
        # This is probably a DER encoded key
        return _import_keyDER(extern_key, passphrase, max_iteration_count, threads, max_memory)

    raise ValueError("RSA key format is not supported")


# Backward compatibility
importKey = import_key

#: `Object ID`_ for the RSA encryption algorithm. This OID often indicates
#: a generic RSA key, even when such key will be actually used for digital
#: signatures.
#:
#: .. note:
#:    An RSA key meant for PSS padding has a dedicated Object ID ``1.2.840.113549.1.1.10``
#:
#: .. _`Object ID`: http://www.alvestrand.no/objectid/1.2.840.113549.1.1.1.html
oid: str = "1.2.840.113549.1.1.1"
