first add files

This commit is contained in:
2023-10-08 20:59:00 +08:00
parent b494be364b
commit 1dac226337
991 changed files with 368151 additions and 40 deletions

682
lib/Crypto/PublicKey/DSA.py Normal file
View File

@@ -0,0 +1,682 @@
# -*- coding: utf-8 -*-
#
# PublicKey/DSA.py : DSA signature primitive
#
# Written in 2008 by Dwayne C. Litzenberger <dlitz@dlitz.net>
#
# ===================================================================
# The contents of this file are dedicated to the public domain. To
# the extent that dedication to the public domain is not available,
# everyone is granted a worldwide, perpetual, royalty-free,
# non-exclusive license to exercise all rights associated with the
# contents of this file for any purpose whatsoever.
# No rights are reserved.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
# ===================================================================
__all__ = ['generate', 'construct', 'DsaKey', 'import_key' ]
import binascii
import struct
import itertools
from Crypto.Util.py3compat import bchr, bord, tobytes, tostr, iter_range
from Crypto import Random
from Crypto.IO import PKCS8, PEM
from Crypto.Hash import SHA256
from Crypto.Util.asn1 import (
DerObject, DerSequence,
DerInteger, DerObjectId,
DerBitString,
)
from Crypto.Math.Numbers import Integer
from Crypto.Math.Primality import (test_probable_prime, COMPOSITE,
PROBABLY_PRIME)
from Crypto.PublicKey import (_expand_subject_public_key_info,
_create_subject_public_key_info,
_extract_subject_public_key_info)
# ; The following ASN.1 types are relevant for DSA
#
# SubjectPublicKeyInfo ::= SEQUENCE {
# algorithm AlgorithmIdentifier,
# subjectPublicKey BIT STRING
# }
#
# id-dsa ID ::= { iso(1) member-body(2) us(840) x9-57(10040) x9cm(4) 1 }
#
# ; See RFC3279
# Dss-Parms ::= SEQUENCE {
# p INTEGER,
# q INTEGER,
# g INTEGER
# }
#
# DSAPublicKey ::= INTEGER
#
# DSSPrivatKey_OpenSSL ::= SEQUENCE
# version INTEGER,
# p INTEGER,
# q INTEGER,
# g INTEGER,
# y INTEGER,
# x INTEGER
# }
#
class DsaKey(object):
r"""Class defining an actual DSA key.
Do not instantiate directly.
Use :func:`generate`, :func:`construct` or :func:`import_key` instead.
:ivar p: DSA modulus
:vartype p: integer
:ivar q: Order of the subgroup
:vartype q: integer
:ivar g: Generator
:vartype g: integer
:ivar y: Public key
:vartype y: integer
:ivar x: Private key
:vartype x: integer
:undocumented: exportKey, publickey
"""
_keydata = ['y', 'g', 'p', 'q', 'x']
def __init__(self, key_dict):
input_set = set(key_dict.keys())
public_set = set(('y' , 'g', 'p', 'q'))
if not public_set.issubset(input_set):
raise ValueError("Some DSA components are missing = %s" %
str(public_set - input_set))
extra_set = input_set - public_set
if extra_set and extra_set != set(('x',)):
raise ValueError("Unknown DSA components = %s" %
str(extra_set - set(('x',))))
self._key = dict(key_dict)
def _sign(self, m, k):
if not self.has_private():
raise TypeError("DSA public key cannot be used for signing")
if not (1 < k < self.q):
raise ValueError("k is not between 2 and q-1")
x, q, p, g = [self._key[comp] for comp in ['x', 'q', 'p', 'g']]
blind_factor = Integer.random_range(min_inclusive=1,
max_exclusive=q)
inv_blind_k = (blind_factor * k).inverse(q)
blind_x = x * blind_factor
r = pow(g, k, p) % q # r = (g**k mod p) mod q
s = (inv_blind_k * (blind_factor * m + blind_x * r)) % q
return map(int, (r, s))
def _verify(self, m, sig):
r, s = sig
y, q, p, g = [self._key[comp] for comp in ['y', 'q', 'p', 'g']]
if not (0 < r < q) or not (0 < s < q):
return False
w = Integer(s).inverse(q)
u1 = (w * m) % q
u2 = (w * r) % q
v = (pow(g, u1, p) * pow(y, u2, p) % p) % q
return v == r
def has_private(self):
"""Whether this is a DSA private key"""
return 'x' in self._key
def can_encrypt(self): # legacy
return False
def can_sign(self): # legacy
return True
def public_key(self):
"""A matching DSA public key.
Returns:
a new :class:`DsaKey` object
"""
public_components = dict((k, self._key[k]) for k in ('y', 'g', 'p', 'q'))
return DsaKey(public_components)
def __eq__(self, other):
if bool(self.has_private()) != bool(other.has_private()):
return False
result = True
for comp in self._keydata:
result = result and (getattr(self._key, comp, None) ==
getattr(other._key, comp, None))
return result
def __ne__(self, other):
return not self.__eq__(other)
def __getstate__(self):
# DSA key is not pickable
from pickle import PicklingError
raise PicklingError
def domain(self):
"""The DSA domain parameters.
Returns
tuple : (p,q,g)
"""
return [int(self._key[comp]) for comp in ('p', 'q', 'g')]
def __repr__(self):
attrs = []
for k in self._keydata:
if k == 'p':
bits = Integer(self.p).size_in_bits()
attrs.append("p(%d)" % (bits,))
elif hasattr(self, k):
attrs.append(k)
if self.has_private():
attrs.append("private")
# PY3K: This is meant to be text, do not change to bytes (data)
return "<%s @0x%x %s>" % (self.__class__.__name__, id(self), ",".join(attrs))
def __getattr__(self, item):
try:
return int(self._key[item])
except KeyError:
raise AttributeError(item)
def export_key(self, format='PEM', pkcs8=None, passphrase=None,
protection=None, randfunc=None):
"""Export this DSA key.
Args:
format (string):
The encoding for the output:
- *'PEM'* (default). ASCII as per `RFC1421`_/ `RFC1423`_.
- *'DER'*. Binary ASN.1 encoding.
- *'OpenSSH'*. ASCII one-liner as per `RFC4253`_.
Only suitable for public keys, not for private keys.
passphrase (string):
*Private keys only*. The pass phrase to protect the output.
pkcs8 (boolean):
*Private keys only*. If ``True`` (default), the key is encoded
with `PKCS#8`_. If ``False``, it is encoded in the custom
OpenSSL/OpenSSH container.
protection (string):
*Only in combination with a pass phrase*.
The encryption scheme to use to protect the output.
If :data:`pkcs8` takes value ``True``, this is the PKCS#8
algorithm to use for deriving the secret and encrypting
the private DSA key.
For a complete list of algorithms, see :mod:`Crypto.IO.PKCS8`.
The default is *PBKDF2WithHMAC-SHA1AndDES-EDE3-CBC*.
If :data:`pkcs8` is ``False``, the obsolete PEM encryption scheme is
used. It is based on MD5 for key derivation, and Triple DES for
encryption. Parameter :data:`protection` is then ignored.
The combination ``format='DER'`` and ``pkcs8=False`` is not allowed
if a passphrase is present.
randfunc (callable):
A function that returns random bytes.
By default it is :func:`Crypto.Random.get_random_bytes`.
Returns:
byte string : the encoded key
Raises:
ValueError : when the format is unknown or when you try to encrypt a private
key with *DER* format and OpenSSL/OpenSSH.
.. warning::
If you don't provide a pass phrase, the private key will be
exported in the clear!
.. _RFC1421: http://www.ietf.org/rfc/rfc1421.txt
.. _RFC1423: http://www.ietf.org/rfc/rfc1423.txt
.. _RFC4253: http://www.ietf.org/rfc/rfc4253.txt
.. _`PKCS#8`: http://www.ietf.org/rfc/rfc5208.txt
"""
if passphrase is not None:
passphrase = tobytes(passphrase)
if randfunc is None:
randfunc = Random.get_random_bytes
if format == 'OpenSSH':
tup1 = [self._key[x].to_bytes() for x in ('p', 'q', 'g', 'y')]
def func(x):
if (bord(x[0]) & 0x80):
return bchr(0) + x
else:
return x
tup2 = [func(x) for x in tup1]
keyparts = [b'ssh-dss'] + tup2
keystring = b''.join(
[struct.pack(">I", len(kp)) + kp for kp in keyparts]
)
return b'ssh-dss ' + binascii.b2a_base64(keystring)[:-1]
# DER format is always used, even in case of PEM, which simply
# encodes it into BASE64.
params = DerSequence([self.p, self.q, self.g])
if self.has_private():
if pkcs8 is None:
pkcs8 = True
if pkcs8:
if not protection:
protection = 'PBKDF2WithHMAC-SHA1AndDES-EDE3-CBC'
private_key = DerInteger(self.x).encode()
binary_key = PKCS8.wrap(
private_key, oid, passphrase,
protection, key_params=params,
randfunc=randfunc
)
if passphrase:
key_type = 'ENCRYPTED PRIVATE'
else:
key_type = 'PRIVATE'
passphrase = None
else:
if format != 'PEM' and passphrase:
raise ValueError("DSA private key cannot be encrypted")
ints = [0, self.p, self.q, self.g, self.y, self.x]
binary_key = DerSequence(ints).encode()
key_type = "DSA PRIVATE"
else:
if pkcs8:
raise ValueError("PKCS#8 is only meaningful for private keys")
binary_key = _create_subject_public_key_info(oid,
DerInteger(self.y), params)
key_type = "PUBLIC"
if format == 'DER':
return binary_key
if format == 'PEM':
pem_str = PEM.encode(
binary_key, key_type + " KEY",
passphrase, randfunc
)
return tobytes(pem_str)
raise ValueError("Unknown key format '%s'. Cannot export the DSA key." % format)
# Backward-compatibility
exportKey = export_key
publickey = public_key
# Methods defined in PyCrypto that we don't support anymore
def sign(self, M, K):
raise NotImplementedError("Use module Crypto.Signature.DSS instead")
def verify(self, M, signature):
raise NotImplementedError("Use module Crypto.Signature.DSS instead")
def encrypt(self, plaintext, K):
raise NotImplementedError
def decrypt(self, ciphertext):
raise NotImplementedError
def blind(self, M, B):
raise NotImplementedError
def unblind(self, M, B):
raise NotImplementedError
def size(self):
raise NotImplementedError
def _generate_domain(L, randfunc):
"""Generate a new set of DSA domain parameters"""
N = { 1024:160, 2048:224, 3072:256 }.get(L)
if N is None:
raise ValueError("Invalid modulus length (%d)" % L)
outlen = SHA256.digest_size * 8
n = (L + outlen - 1) // outlen - 1 # ceil(L/outlen) -1
b_ = L - 1 - (n * outlen)
# Generate q (A.1.1.2)
q = Integer(4)
upper_bit = 1 << (N - 1)
while test_probable_prime(q, randfunc) != PROBABLY_PRIME:
seed = randfunc(64)
U = Integer.from_bytes(SHA256.new(seed).digest()) & (upper_bit - 1)
q = U | upper_bit | 1
assert(q.size_in_bits() == N)
# Generate p (A.1.1.2)
offset = 1
upper_bit = 1 << (L - 1)
while True:
V = [ SHA256.new(seed + Integer(offset + j).to_bytes()).digest()
for j in iter_range(n + 1) ]
V = [ Integer.from_bytes(v) for v in V ]
W = sum([V[i] * (1 << (i * outlen)) for i in iter_range(n)],
(V[n] & ((1 << b_) - 1)) * (1 << (n * outlen)))
X = Integer(W + upper_bit) # 2^{L-1} < X < 2^{L}
assert(X.size_in_bits() == L)
c = X % (q * 2)
p = X - (c - 1) # 2q divides (p-1)
if p.size_in_bits() == L and \
test_probable_prime(p, randfunc) == PROBABLY_PRIME:
break
offset += n + 1
# Generate g (A.2.3, index=1)
e = (p - 1) // q
for count in itertools.count(1):
U = seed + b"ggen" + bchr(1) + Integer(count).to_bytes()
W = Integer.from_bytes(SHA256.new(U).digest())
g = pow(W, e, p)
if g != 1:
break
return (p, q, g, seed)
def generate(bits, randfunc=None, domain=None):
"""Generate a new DSA key pair.
The algorithm follows Appendix A.1/A.2 and B.1 of `FIPS 186-4`_,
respectively for domain generation and key pair generation.
Args:
bits (integer):
Key length, or size (in bits) of the DSA modulus *p*.
It must be 1024, 2048 or 3072.
randfunc (callable):
Random number generation function; it accepts a single integer N
and return a string of random data N bytes long.
If not specified, :func:`Crypto.Random.get_random_bytes` is used.
domain (tuple):
The DSA domain parameters *p*, *q* and *g* as a list of 3
integers. Size of *p* and *q* must comply to `FIPS 186-4`_.
If not specified, the parameters are created anew.
Returns:
:class:`DsaKey` : a new DSA key object
Raises:
ValueError : when **bits** is too little, too big, or not a multiple of 64.
.. _FIPS 186-4: http://nvlpubs.nist.gov/nistpubs/FIPS/NIST.FIPS.186-4.pdf
"""
if randfunc is None:
randfunc = Random.get_random_bytes
if domain:
p, q, g = map(Integer, domain)
## Perform consistency check on domain parameters
# P and Q must be prime
fmt_error = test_probable_prime(p) == COMPOSITE
fmt_error |= test_probable_prime(q) == COMPOSITE
# Verify Lagrange's theorem for sub-group
fmt_error |= ((p - 1) % q) != 0
fmt_error |= g <= 1 or g >= p
fmt_error |= pow(g, q, p) != 1
if fmt_error:
raise ValueError("Invalid DSA domain parameters")
else:
p, q, g, _ = _generate_domain(bits, randfunc)
L = p.size_in_bits()
N = q.size_in_bits()
if L != bits:
raise ValueError("Mismatch between size of modulus (%d)"
" and 'bits' parameter (%d)" % (L, bits))
if (L, N) not in [(1024, 160), (2048, 224),
(2048, 256), (3072, 256)]:
raise ValueError("Lengths of p and q (%d, %d) are not compatible"
"to FIPS 186-3" % (L, N))
if not 1 < g < p:
raise ValueError("Incorrent DSA generator")
# B.1.1
c = Integer.random(exact_bits=N + 64, randfunc=randfunc)
x = c % (q - 1) + 1 # 1 <= x <= q-1
y = pow(g, x, p)
key_dict = { 'y':y, 'g':g, 'p':p, 'q':q, 'x':x }
return DsaKey(key_dict)
def construct(tup, consistency_check=True):
"""Construct a DSA key from a tuple of valid DSA components.
Args:
tup (tuple):
A tuple of long integers, with 4 or 5 items
in the following order:
1. Public key (*y*).
2. Sub-group generator (*g*).
3. Modulus, finite field order (*p*).
4. Sub-group order (*q*).
5. Private key (*x*). Optional.
consistency_check (boolean):
If ``True``, the library will verify that the provided components
fulfil the main DSA properties.
Raises:
ValueError: when the key being imported fails the most basic DSA validity checks.
Returns:
:class:`DsaKey` : a DSA key object
"""
key_dict = dict(zip(('y', 'g', 'p', 'q', 'x'), map(Integer, tup)))
key = DsaKey(key_dict)
fmt_error = False
if consistency_check:
# P and Q must be prime
fmt_error = test_probable_prime(key.p) == COMPOSITE
fmt_error |= test_probable_prime(key.q) == COMPOSITE
# Verify Lagrange's theorem for sub-group
fmt_error |= ((key.p - 1) % key.q) != 0
fmt_error |= key.g <= 1 or key.g >= key.p
fmt_error |= pow(key.g, key.q, key.p) != 1
# Public key
fmt_error |= key.y <= 0 or key.y >= key.p
if hasattr(key, 'x'):
fmt_error |= key.x <= 0 or key.x >= key.q
fmt_error |= pow(key.g, key.x, key.p) != key.y
if fmt_error:
raise ValueError("Invalid DSA key components")
return key
# Dss-Parms ::= SEQUENCE {
# p OCTET STRING,
# q OCTET STRING,
# g OCTET STRING
# }
# DSAPublicKey ::= INTEGER -- public key, y
def _import_openssl_private(encoded, passphrase, params):
if params:
raise ValueError("DSA private key already comes with parameters")
der = DerSequence().decode(encoded, nr_elements=6, only_ints_expected=True)
if der[0] != 0:
raise ValueError("No version found")
tup = [der[comp] for comp in (4, 3, 1, 2, 5)]
return construct(tup)
def _import_subjectPublicKeyInfo(encoded, passphrase, params):
algoid, encoded_key, emb_params = _expand_subject_public_key_info(encoded)
if algoid != oid:
raise ValueError("No DSA subjectPublicKeyInfo")
if params and emb_params:
raise ValueError("Too many DSA parameters")
y = DerInteger().decode(encoded_key).value
p, q, g = list(DerSequence().decode(params or emb_params))
tup = (y, g, p, q)
return construct(tup)
def _import_x509_cert(encoded, passphrase, params):
sp_info = _extract_subject_public_key_info(encoded)
return _import_subjectPublicKeyInfo(sp_info, None, params)
def _import_pkcs8(encoded, passphrase, params):
if params:
raise ValueError("PKCS#8 already includes parameters")
k = PKCS8.unwrap(encoded, passphrase)
if k[0] != oid:
raise ValueError("No PKCS#8 encoded DSA key")
x = DerInteger().decode(k[1]).value
p, q, g = list(DerSequence().decode(k[2]))
tup = (pow(g, x, p), g, p, q, x)
return construct(tup)
def _import_key_der(key_data, passphrase, params):
"""Import a DSA key (public or private half), encoded in DER form."""
decodings = (_import_openssl_private,
_import_subjectPublicKeyInfo,
_import_x509_cert,
_import_pkcs8)
for decoding in decodings:
try:
return decoding(key_data, passphrase, params)
except ValueError:
pass
raise ValueError("DSA key format is not supported")
def import_key(extern_key, passphrase=None):
"""Import a DSA key.
Args:
extern_key (string or byte string):
The DSA key to import.
The following formats are supported for a DSA **public** key:
- X.509 certificate (binary DER or PEM)
- X.509 ``subjectPublicKeyInfo`` (binary DER or PEM)
- OpenSSH (ASCII one-liner, see `RFC4253`_)
The following formats are supported for a DSA **private** key:
- `PKCS#8`_ ``PrivateKeyInfo`` or ``EncryptedPrivateKeyInfo``
DER SEQUENCE (binary or PEM)
- OpenSSL/OpenSSH custom format (binary or PEM)
For details about the PEM encoding, see `RFC1421`_/`RFC1423`_.
passphrase (string):
In case of an encrypted private key, this is the pass phrase
from which the decryption key is derived.
Encryption may be applied either at the `PKCS#8`_ or at the PEM level.
Returns:
:class:`DsaKey` : a DSA key object
Raises:
ValueError : when the given key cannot be parsed (possibly because
the pass phrase is wrong).
.. _RFC1421: http://www.ietf.org/rfc/rfc1421.txt
.. _RFC1423: http://www.ietf.org/rfc/rfc1423.txt
.. _RFC4253: http://www.ietf.org/rfc/rfc4253.txt
.. _PKCS#8: http://www.ietf.org/rfc/rfc5208.txt
"""
extern_key = tobytes(extern_key)
if passphrase is not None:
passphrase = tobytes(passphrase)
if extern_key.startswith(b'-----'):
# This is probably a PEM encoded key
(der, marker, enc_flag) = PEM.decode(tostr(extern_key), passphrase)
if enc_flag:
passphrase = None
return _import_key_der(der, passphrase, None)
if extern_key.startswith(b'ssh-dss '):
# This is probably a public 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:]
if keyparts[0] == b"ssh-dss":
tup = [Integer.from_bytes(keyparts[x]) for x in (4, 3, 1, 2)]
return construct(tup)
if len(extern_key) > 0 and bord(extern_key[0]) == 0x30:
# This is probably a DER encoded key
return _import_key_der(extern_key, passphrase, None)
raise ValueError("DSA key format is not supported")
# Backward compatibility
importKey = import_key
#: `Object ID`_ for a DSA key.
#:
#: id-dsa ID ::= { iso(1) member-body(2) us(840) x9-57(10040) x9cm(4) 1 }
#:
#: .. _`Object ID`: http://www.alvestrand.no/objectid/1.2.840.10040.4.1.html
oid = "1.2.840.10040.4.1"

View File

@@ -0,0 +1,31 @@
from typing import Dict, Tuple, Callable, Union, Optional
__all__ = ['generate', 'construct', 'DsaKey', 'import_key' ]
RNG = Callable[[int], bytes]
class DsaKey(object):
def __init__(self, key_dict: Dict[str, int]) -> None: ...
def has_private(self) -> bool: ...
def can_encrypt(self) -> bool: ... # legacy
def can_sign(self) -> bool: ... # legacy
def public_key(self) -> DsaKey: ...
def __eq__(self, other: object) -> bool: ...
def __ne__(self, other: object) -> bool: ...
def __getstate__(self) -> None: ...
def domain(self) -> Tuple[int, int, int]: ...
def __repr__(self) -> str: ...
def __getattr__(self, item: str) -> int: ...
def export_key(self, format: Optional[str]="PEM", pkcs8: Optional[bool]=None, passphrase: Optional[str]=None,
protection: Optional[str]=None, randfunc: Optional[RNG]=None) -> bytes: ...
# Backward-compatibility
exportKey = export_key
publickey = public_key
def generate(bits: int, randfunc: Optional[RNG]=None, domain: Optional[Tuple[int, int, int]]=None) -> DsaKey: ...
def construct(tup: Union[Tuple[int, int, int, int], Tuple[int, int, int, int, int]], consistency_check: Optional[bool]=True) -> DsaKey: ...
def import_key(extern_key: Union[str, bytes], passphrase: Optional[str]=None) -> DsaKey: ...
# Backward compatibility
importKey = import_key
oid: str

1794
lib/Crypto/PublicKey/ECC.py Normal file

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,66 @@
from typing import Union, Callable, Optional, NamedTuple, List, Tuple, Dict, NamedTuple, Any
from Crypto.Math.Numbers import Integer
RNG = Callable[[int], bytes]
class UnsupportedEccFeature(ValueError): ...
class EccPoint(object):
def __init__(self, x: Union[int, Integer], y: Union[int, Integer], curve: Optional[str] = ...) -> None: ...
def set(self, point: EccPoint) -> EccPoint: ...
def __eq__(self, point: object) -> bool: ...
def __neg__(self) -> EccPoint: ...
def copy(self) -> EccPoint: ...
def is_point_at_infinity(self) -> bool: ...
def point_at_infinity(self) -> EccPoint: ...
@property
def x(self) -> int: ...
@property
def y(self) -> int: ...
@property
def xy(self) -> Tuple[int, int]: ...
def size_in_bytes(self) -> int: ...
def size_in_bits(self) -> int: ...
def double(self) -> EccPoint: ...
def __iadd__(self, point: EccPoint) -> EccPoint: ...
def __add__(self, point: EccPoint) -> EccPoint: ...
def __imul__(self, scalar: int) -> EccPoint: ...
def __mul__(self, scalar: int) -> EccPoint: ...
class EccKey(object):
curve: str
def __init__(self, *, curve: str = ..., d: int = ..., point: EccPoint = ...) -> None: ...
def __eq__(self, other: object) -> bool: ...
def __repr__(self) -> str: ...
def has_private(self) -> bool: ...
@property
def d(self) -> int: ...
@property
def pointQ(self) -> EccPoint: ...
def public_key(self) -> EccKey: ...
def export_key(self, **kwargs: Union[str, bytes, bool]) -> Union[str,bytes]: ...
_Curve = NamedTuple("_Curve", [('p', Integer),
('order', Integer),
('b', Integer),
('Gx', Integer),
('Gy', Integer),
('G', EccPoint),
('modulus_bits', int),
('oid', str),
('context', Any),
('desc', str),
('openssh', Union[str, None]),
])
_curves : Dict[str, _Curve]
def generate(**kwargs: Union[str, RNG]) -> EccKey: ...
def construct(**kwargs: Union[str, int]) -> EccKey: ...
def import_key(encoded: Union[bytes, str],
passphrase: Optional[str]=None,
curve_name:Optional[str]=None) -> EccKey: ...
def _import_ed25519_public_key(encoded: bytes) -> EccKey: ...
def _import_ed448_public_key(encoded: bytes) -> EccKey: ...

View File

@@ -0,0 +1,286 @@
#
# ElGamal.py : ElGamal encryption/decryption and signatures
#
# Part of the Python Cryptography Toolkit
#
# Originally written by: A.M. Kuchling
#
# ===================================================================
# The contents of this file are dedicated to the public domain. To
# the extent that dedication to the public domain is not available,
# everyone is granted a worldwide, perpetual, royalty-free,
# non-exclusive license to exercise all rights associated with the
# contents of this file for any purpose whatsoever.
# No rights are reserved.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
# ===================================================================
__all__ = ['generate', 'construct', 'ElGamalKey']
from Crypto import Random
from Crypto.Math.Primality import ( generate_probable_safe_prime,
test_probable_prime, COMPOSITE )
from Crypto.Math.Numbers import Integer
# Generate an ElGamal key with N bits
def generate(bits, randfunc):
"""Randomly generate a fresh, new ElGamal key.
The key will be safe for use for both encryption and signature
(although it should be used for **only one** purpose).
Args:
bits (int):
Key length, or size (in bits) of the modulus *p*.
The recommended value is 2048.
randfunc (callable):
Random number generation function; it should accept
a single integer *N* and return a string of random
*N* random bytes.
Return:
an :class:`ElGamalKey` object
"""
obj=ElGamalKey()
# Generate a safe prime p
# See Algorithm 4.86 in Handbook of Applied Cryptography
obj.p = generate_probable_safe_prime(exact_bits=bits, randfunc=randfunc)
q = (obj.p - 1) >> 1
# Generate generator g
while 1:
# Choose a square residue; it will generate a cyclic group of order q.
obj.g = pow(Integer.random_range(min_inclusive=2,
max_exclusive=obj.p,
randfunc=randfunc), 2, obj.p)
# We must avoid g=2 because of Bleichenbacher's attack described
# in "Generating ElGamal signatures without knowning the secret key",
# 1996
if obj.g in (1, 2):
continue
# Discard g if it divides p-1 because of the attack described
# in Note 11.67 (iii) in HAC
if (obj.p - 1) % obj.g == 0:
continue
# g^{-1} must not divide p-1 because of Khadir's attack
# described in "Conditions of the generator for forging ElGamal
# signature", 2011
ginv = obj.g.inverse(obj.p)
if (obj.p - 1) % ginv == 0:
continue
# Found
break
# Generate private key x
obj.x = Integer.random_range(min_inclusive=2,
max_exclusive=obj.p-1,
randfunc=randfunc)
# Generate public key y
obj.y = pow(obj.g, obj.x, obj.p)
return obj
def construct(tup):
r"""Construct an ElGamal key from a tuple of valid ElGamal components.
The modulus *p* must be a prime.
The following conditions must apply:
.. math::
\begin{align}
&1 < g < p-1 \\
&g^{p-1} = 1 \text{ mod } 1 \\
&1 < x < p-1 \\
&g^x = y \text{ mod } p
\end{align}
Args:
tup (tuple):
A tuple with either 3 or 4 integers,
in the following order:
1. Modulus (*p*).
2. Generator (*g*).
3. Public key (*y*).
4. Private key (*x*). Optional.
Raises:
ValueError: when the key being imported fails the most basic ElGamal validity checks.
Returns:
an :class:`ElGamalKey` object
"""
obj=ElGamalKey()
if len(tup) not in [3,4]:
raise ValueError('argument for construct() wrong length')
for i in range(len(tup)):
field = obj._keydata[i]
setattr(obj, field, Integer(tup[i]))
fmt_error = test_probable_prime(obj.p) == COMPOSITE
fmt_error |= obj.g<=1 or obj.g>=obj.p
fmt_error |= pow(obj.g, obj.p-1, obj.p)!=1
fmt_error |= obj.y<1 or obj.y>=obj.p
if len(tup)==4:
fmt_error |= obj.x<=1 or obj.x>=obj.p
fmt_error |= pow(obj.g, obj.x, obj.p)!=obj.y
if fmt_error:
raise ValueError("Invalid ElGamal key components")
return obj
class ElGamalKey(object):
r"""Class defining an ElGamal key.
Do not instantiate directly.
Use :func:`generate` or :func:`construct` instead.
:ivar p: Modulus
:vartype d: integer
:ivar g: Generator
:vartype e: integer
:ivar y: Public key component
:vartype y: integer
:ivar x: Private key component
:vartype x: integer
"""
#: Dictionary of ElGamal parameters.
#:
#: A public key will only have the following entries:
#:
#: - **y**, the public key.
#: - **g**, the generator.
#: - **p**, the modulus.
#:
#: A private key will also have:
#:
#: - **x**, the private key.
_keydata=['p', 'g', 'y', 'x']
def __init__(self, randfunc=None):
if randfunc is None:
randfunc = Random.new().read
self._randfunc = randfunc
def _encrypt(self, M, K):
a=pow(self.g, K, self.p)
b=( pow(self.y, K, self.p)*M ) % self.p
return [int(a), int(b)]
def _decrypt(self, M):
if (not hasattr(self, 'x')):
raise TypeError('Private key not available in this object')
r = Integer.random_range(min_inclusive=2,
max_exclusive=self.p-1,
randfunc=self._randfunc)
a_blind = (pow(self.g, r, self.p) * M[0]) % self.p
ax=pow(a_blind, self.x, self.p)
plaintext_blind = (ax.inverse(self.p) * M[1] ) % self.p
plaintext = (plaintext_blind * pow(self.y, r, self.p)) % self.p
return int(plaintext)
def _sign(self, M, K):
if (not hasattr(self, 'x')):
raise TypeError('Private key not available in this object')
p1=self.p-1
K = Integer(K)
if (K.gcd(p1)!=1):
raise ValueError('Bad K value: GCD(K,p-1)!=1')
a=pow(self.g, K, self.p)
t=(Integer(M)-self.x*a) % p1
while t<0: t=t+p1
b=(t*K.inverse(p1)) % p1
return [int(a), int(b)]
def _verify(self, M, sig):
sig = [Integer(x) for x in sig]
if sig[0]<1 or sig[0]>self.p-1:
return 0
v1=pow(self.y, sig[0], self.p)
v1=(v1*pow(sig[0], sig[1], self.p)) % self.p
v2=pow(self.g, M, self.p)
if v1==v2:
return 1
return 0
def has_private(self):
"""Whether this is an ElGamal private key"""
if hasattr(self, 'x'):
return 1
else:
return 0
def can_encrypt(self):
return True
def can_sign(self):
return True
def publickey(self):
"""A matching ElGamal public key.
Returns:
a new :class:`ElGamalKey` object
"""
return construct((self.p, self.g, self.y))
def __eq__(self, other):
if bool(self.has_private()) != bool(other.has_private()):
return False
result = True
for comp in self._keydata:
result = result and (getattr(self.key, comp, None) ==
getattr(other.key, comp, None))
return result
def __ne__(self, other):
return not self.__eq__(other)
def __getstate__(self):
# ElGamal key is not pickable
from pickle import PicklingError
raise PicklingError
# Methods defined in PyCrypto that we don't support anymore
def sign(self, M, K):
raise NotImplementedError
def verify(self, M, signature):
raise NotImplementedError
def encrypt(self, plaintext, K):
raise NotImplementedError
def decrypt(self, ciphertext):
raise NotImplementedError
def blind(self, M, B):
raise NotImplementedError
def unblind(self, M, B):
raise NotImplementedError
def size(self):
raise NotImplementedError

View File

@@ -0,0 +1,18 @@
from typing import Callable, Union, Tuple, Optional
__all__ = ['generate', 'construct', 'ElGamalKey']
RNG = Callable[[int], bytes]
def generate(bits: int, randfunc: RNG) -> ElGamalKey: ...
def construct(tup: Union[Tuple[int, int, int], Tuple[int, int, int, int]]) -> ElGamalKey: ...
class ElGamalKey(object):
def __init__(self, randfunc: Optional[RNG]=None) -> None: ...
def has_private(self) -> bool: ...
def can_encrypt(self) -> bool: ...
def can_sign(self) -> bool: ...
def publickey(self) -> ElGamalKey: ...
def __eq__(self, other: object) -> bool: ...
def __ne__(self, other: object) -> bool: ...
def __getstate__(self) -> None: ...

802
lib/Crypto/PublicKey/RSA.py Normal file
View File

@@ -0,0 +1,802 @@
# -*- coding: utf-8 -*-
# ===================================================================
#
# 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.
# ===================================================================
__all__ = ['generate', 'construct', 'import_key',
'RsaKey', 'oid']
import binascii
import struct
from Crypto import Random
from Crypto.Util.py3compat import tobytes, bord, tostr
from Crypto.Util.asn1 import DerSequence, DerNull
from Crypto.Math.Numbers import Integer
from Crypto.Math.Primality import (test_probable_prime,
generate_probable_prime, COMPOSITE)
from Crypto.PublicKey import (_expand_subject_public_key_info,
_create_subject_public_key_info,
_extract_subject_public_key_info)
class RsaKey(object):
r"""Class defining an actual RSA key.
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 u: Chinese remainder component (:math:`p^{-1} \text{mod } q`)
:vartype u: integer
:undocumented: exportKey, publickey
"""
def __init__(self, **kwargs):
"""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 = set(('n', 'e'))
private_set = public_set | 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)
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)
@property
def n(self):
return int(self._n)
@property
def e(self):
return int(self._e)
@property
def d(self):
if not self.has_private():
raise AttributeError("No private exponent available for public keys")
return int(self._d)
@property
def p(self):
if not self.has_private():
raise AttributeError("No CRT component 'p' available for public keys")
return int(self._p)
@property
def q(self):
if not self.has_private():
raise AttributeError("No CRT component 'q' available for public keys")
return int(self._q)
@property
def u(self):
if not self.has_private():
raise AttributeError("No CRT component 'u' available for public keys")
return int(self._u)
def size_in_bits(self):
"""Size of the RSA modulus in bits"""
return self._n.size_in_bits()
def size_in_bytes(self):
"""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(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: Generate random secret blinding factor r,
# such that 0 < r < n-1
r = Integer.random_range(min_inclusive=1, max_exclusive=self._n)
# Step 2: Compute c' = c * r**e mod n
cp = Integer(ciphertext) * pow(r, self._e, self._n) % self._n
# Step 3: Compute m' = c'**d mod n (normal RSA decryption)
m1 = pow(cp, self._dp, self._p)
m2 = pow(cp, self._dq, self._q)
h = ((m2 - m1) * self._u) % self._q
mp = h * self._p + m1
# Step 4: Compute m = m' * (r**(-1)) mod n
result = (r.inverse(self._n) * mp) % self._n
# Verify no faults occurred
if ciphertext != pow(result, self._e, self._n):
raise ValueError("Fault detected in RSA decryption")
return result
def has_private(self):
"""Whether this is an RSA private key"""
return hasattr(self, "_d")
def can_encrypt(self): # legacy
return True
def can_sign(self): # legacy
return True
def public_key(self):
"""A matching RSA public key.
Returns:
a new :class:`RsaKey` object
"""
return RsaKey(n=self._n, e=self._e)
def __eq__(self, other):
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 __ne__(self, other):
return not (self == other)
def __getstate__(self):
# RSA key is not pickable
from pickle import PicklingError
raise PicklingError
def __repr__(self):
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):
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='PEM', passphrase=None, pkcs=1,
protection=None, randfunc=None):
"""Export this RSA key.
Args:
format (string):
The format to use for wrapping the key:
- *'PEM'*. (*Default*) Text encoding, done according to `RFC1421`_/`RFC1423`_.
- *'DER'*. Binary encoding.
- *'OpenSSH'*. Textual encoding, done according to OpenSSH specification.
Only suitable for public keys (not private keys).
passphrase (string):
(*For private keys only*) The pass phrase used for protecting the output.
pkcs (integer):
(*For private keys only*) The ASN.1 structure to use for
serializing the key. Note that even in case of PEM
encoding, there is an inner ASN.1 DER structure.
With ``pkcs=1`` (*default*), the private key is encoded in a
simple `PKCS#1`_ structure (``RSAPrivateKey``).
With ``pkcs=8``, the private key is encoded in a `PKCS#8`_ structure
(``PrivateKeyInfo``).
.. note::
This parameter is ignored for a public key.
For DER and PEM, an ASN.1 DER ``SubjectPublicKeyInfo``
structure is always used.
protection (string):
(*For private keys only*)
The encryption scheme to use for protecting the private key.
If ``None`` (default), the behavior depends on :attr:`format`:
- For *'DER'*, the *PBKDF2WithHMAC-SHA1AndDES-EDE3-CBC*
scheme is used. The following operations are performed:
1. A 16 byte Triple DES key is derived from the passphrase
using :func:`Crypto.Protocol.KDF.PBKDF2` with 8 bytes salt,
and 1 000 iterations of :mod:`Crypto.Hash.HMAC`.
2. The private key is encrypted using CBC.
3. The encrypted key is encoded according to PKCS#8.
- For *'PEM'*, the obsolete PEM encryption scheme is used.
It is based on MD5 for key derivation, and Triple DES for encryption.
Specifying a value for :attr:`protection` is only meaningful for PKCS#8
(that is, ``pkcs=8``) and only if a pass phrase is present too.
The supported schemes for PKCS#8 are listed in the
:mod:`Crypto.IO.PKCS8` module (see :attr:`wrap_algo` parameter).
randfunc (callable):
A function that provides random bytes. Only used for PEM encoding.
The default is :func:`Crypto.Random.get_random_bytes`.
Returns:
byte string: the encoded key
Raises:
ValueError:when the format is unknown or when you try to encrypt a private
key with *DER* format and PKCS#1.
.. warning::
If you don't provide a pass phrase, the private key will be
exported in the clear!
.. _RFC1421: http://www.ietf.org/rfc/rfc1421.txt
.. _RFC1423: http://www.ietf.org/rfc/rfc1423.txt
.. _`PKCS#1`: http://www.ietf.org/rfc/rfc3447.txt
.. _`PKCS#8`: http://www.ietf.org/rfc/rfc5208.txt
"""
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 bord(e_bytes[0]) & 0x80:
e_bytes = b'\x00' + e_bytes
if bord(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():
binary_key = DerSequence([0,
self.n,
self.e,
self.d,
self.p,
self.q,
self.d % (self.p-1),
self.d % (self.q-1),
Integer(self.q).inverse(self.p)
]).encode()
if pkcs == 1:
key_type = 'RSA PRIVATE KEY'
if format == 'DER' and passphrase:
raise ValueError("PKCS#1 private key cannot be encrypted")
else: # PKCS#8
from Crypto.IO import PKCS8
if format == 'PEM' and protection is None:
key_type = 'PRIVATE KEY'
binary_key = PKCS8.wrap(binary_key, oid, None,
key_params=DerNull())
else:
key_type = 'ENCRYPTED PRIVATE KEY'
if not protection:
protection = 'PBKDF2WithHMAC-SHA1AndDES-EDE3-CBC'
binary_key = PKCS8.wrap(binary_key, oid,
passphrase, protection,
key_params=DerNull())
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
exportKey = export_key
publickey = public_key
# Methods defined in PyCrypto that we don't support anymore
def sign(self, M, K):
raise NotImplementedError("Use module Crypto.Signature.pkcs1_15 instead")
def verify(self, M, signature):
raise NotImplementedError("Use module Crypto.Signature.pkcs1_15 instead")
def encrypt(self, plaintext, K):
raise NotImplementedError("Use module Crypto.Cipher.PKCS1_OAEP instead")
def decrypt(self, ciphertext):
raise NotImplementedError("Use module Crypto.Cipher.PKCS1_OAEP instead")
def blind(self, M, B):
raise NotImplementedError
def unblind(self, M, B):
raise NotImplementedError
def size(self):
raise NotImplementedError
def generate(bits, randfunc=None, e=65537):
"""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.
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 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 and 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
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 and
(candidate - 1).gcd(e) == 1 and
abs(candidate - p) > min_distance)
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, consistency_check=True):
r"""Construct an RSA key from a tuple of valid RSA components.
The modulus **n** must be the product of two primes.
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 *q*, that is :math:`p^{-1} \text{mod }q`. Optional.
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`).
"""
class InputComps(object):
pass
input_comps = InputComps()
for (comp, value) in zip(('n', 'e', 'd', 'p', 'q', 'u'), rsa_components):
setattr(input_comps, comp, Integer(value))
n = input_comps.n
e = input_comps.e
if not hasattr(input_comps, 'd'):
key = RsaKey(n=n, e=e)
else:
d = input_comps.d
if hasattr(input_comps, 'q'):
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 hasattr(input_comps, 'u'):
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 Integer(n).gcd(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 % int(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):
algoid, encoded_key, params = _expand_subject_public_key_info(encoded)
if algoid != oid 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):
from Crypto.IO import PKCS8
k = PKCS8.unwrap(encoded, passphrase)
if k[0] != oid:
raise ValueError("No PKCS#8 encoded RSA key")
return _import_keyDER(k[1], passphrase)
def _import_keyDER(extern_key, passphrase):
"""Import an RSA key (public or private half), encoded in DER form."""
decodings = (_import_pkcs1_private,
_import_pkcs1_public,
_import_subjectPublicKeyInfo,
_import_x509_cert,
_import_pkcs8)
for decoding in decodings:
try:
return decoding(extern_key, passphrase)
except ValueError:
pass
raise ValueError("RSA key format is not supported")
def _import_openssh_private_rsa(data, password):
from ._openssh import (import_openssh_private_generic,
read_bytes, read_string, check_padding)
ssh_name, decrypted = import_openssh_private_generic(data, password)
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, passphrase=None):
"""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.
Returns: An RSA key object (:class:`RsaKey`).
Raises:
ValueError/IndexError/TypeError:
When the given key cannot be parsed (possibly because the pass
phrase is wrong).
.. _RFC1421: http://www.ietf.org/rfc/rfc1421.txt
.. _RFC1423: http://www.ietf.org/rfc/rfc1423.txt
.. _`PKCS#1`: http://www.ietf.org/rfc/rfc3447.txt
.. _`PKCS#8`: http://www.ietf.org/rfc/rfc5208.txt
.. _`OpenSSH 6.5`: https://flak.tedunangst.com/post/new-openssh-key-format-and-bcrypt-pbkdf
"""
from Crypto.IO import PEM
extern_key = tobytes(extern_key)
if passphrase is not None:
passphrase = tobytes(passphrase)
if extern_key.startswith(b'-----BEGIN OPENSSH PRIVATE KEY'):
text_encoded = tostr(extern_key)
openssh_encoded, marker, enc_flag = PEM.decode(text_encoded, passphrase)
result = _import_openssh_private_rsa(openssh_encoded, passphrase)
return result
if extern_key.startswith(b'-----'):
# This is probably a PEM encoded key.
(der, marker, enc_flag) = PEM.decode(tostr(extern_key), passphrase)
if enc_flag:
passphrase = None
return _import_keyDER(der, passphrase)
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 bord(extern_key[0]) == 0x30:
# This is probably a DER encoded key
return _import_keyDER(extern_key, passphrase)
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.
#:
#: .. _`Object ID`: http://www.alvestrand.no/objectid/1.2.840.113549.1.1.1.html
oid = "1.2.840.113549.1.1.1"

View File

@@ -0,0 +1,51 @@
from typing import Callable, Union, Tuple, Optional
__all__ = ['generate', 'construct', 'import_key',
'RsaKey', 'oid']
RNG = Callable[[int], bytes]
class RsaKey(object):
def __init__(self, **kwargs: int) -> None: ...
@property
def n(self) -> int: ...
@property
def e(self) -> int: ...
@property
def d(self) -> int: ...
@property
def p(self) -> int: ...
@property
def q(self) -> int: ...
@property
def u(self) -> int: ...
def size_in_bits(self) -> int: ...
def size_in_bytes(self) -> int: ...
def has_private(self) -> bool: ...
def can_encrypt(self) -> bool: ... # legacy
def can_sign(self) -> bool:... # legacy
def public_key(self) -> RsaKey: ...
def __eq__(self, other: object) -> bool: ...
def __ne__(self, other: object) -> bool: ...
def __getstate__(self) -> None: ...
def __repr__(self) -> str: ...
def __str__(self) -> str: ...
def export_key(self, format: Optional[str]="PEM", passphrase: Optional[str]=None, pkcs: Optional[int]=1,
protection: Optional[str]=None, randfunc: Optional[RNG]=None) -> bytes: ...
# Backward compatibility
exportKey = export_key
publickey = public_key
def generate(bits: int, randfunc: Optional[RNG]=None, e: Optional[int]=65537) -> RsaKey: ...
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
consistency_check: Optional[bool]=True) -> RsaKey: ...
def import_key(extern_key: Union[str, bytes], passphrase: Optional[str]=None) -> RsaKey: ...
# Backward compatibility
importKey = import_key
oid: str

View File

@@ -0,0 +1,94 @@
# -*- coding: utf-8 -*-
#
# ===================================================================
# The contents of this file are dedicated to the public domain. To
# the extent that dedication to the public domain is not available,
# everyone is granted a worldwide, perpetual, royalty-free,
# non-exclusive license to exercise all rights associated with the
# contents of this file for any purpose whatsoever.
# No rights are reserved.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
# ===================================================================
from Crypto.Util.asn1 import (DerSequence, DerInteger, DerBitString,
DerObjectId, DerNull)
def _expand_subject_public_key_info(encoded):
"""Parse a SubjectPublicKeyInfo structure.
It returns a triple with:
* OID (string)
* encoded public key (bytes)
* Algorithm parameters (bytes or None)
"""
#
# SubjectPublicKeyInfo ::= SEQUENCE {
# algorithm AlgorithmIdentifier,
# subjectPublicKey BIT STRING
# }
#
# AlgorithmIdentifier ::= SEQUENCE {
# algorithm OBJECT IDENTIFIER,
# parameters ANY DEFINED BY algorithm OPTIONAL
# }
#
spki = DerSequence().decode(encoded, nr_elements=2)
algo = DerSequence().decode(spki[0], nr_elements=(1,2))
algo_oid = DerObjectId().decode(algo[0])
spk = DerBitString().decode(spki[1]).value
if len(algo) == 1:
algo_params = None
else:
try:
DerNull().decode(algo[1])
algo_params = None
except:
algo_params = algo[1]
return algo_oid.value, spk, algo_params
def _create_subject_public_key_info(algo_oid, public_key, params):
if params is None:
algorithm = DerSequence([DerObjectId(algo_oid)])
else:
algorithm = DerSequence([DerObjectId(algo_oid), params])
spki = DerSequence([algorithm,
DerBitString(public_key)
])
return spki.encode()
def _extract_subject_public_key_info(x509_certificate):
"""Extract subjectPublicKeyInfo from a DER X.509 certificate."""
certificate = DerSequence().decode(x509_certificate, nr_elements=3)
tbs_certificate = DerSequence().decode(certificate[0],
nr_elements=range(6, 11))
index = 5
try:
tbs_certificate[0] + 1
# Version not present
version = 1
except TypeError:
version = DerInteger(explicit=0).decode(tbs_certificate[0]).value
if version not in (2, 3):
raise ValueError("Incorrect X.509 certificate version")
index = 6
return tbs_certificate[index]

View File

Binary file not shown.

Binary file not shown.

Binary file not shown.

View File

@@ -0,0 +1,135 @@
# ===================================================================
#
# Copyright (c) 2019, Helder Eijs <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.
# ===================================================================
import struct
from Crypto.Cipher import AES
from Crypto.Hash import SHA512
from Crypto.Protocol.KDF import _bcrypt_hash
from Crypto.Util.strxor import strxor
from Crypto.Util.py3compat import tostr, bchr, bord
def read_int4(data):
if len(data) < 4:
raise ValueError("Insufficient data")
value = struct.unpack(">I", data[:4])[0]
return value, data[4:]
def read_bytes(data):
size, data = read_int4(data)
if len(data) < size:
raise ValueError("Insufficient data (V)")
return data[:size], data[size:]
def read_string(data):
s, d = read_bytes(data)
return tostr(s), d
def check_padding(pad):
for v, x in enumerate(pad):
if bord(x) != ((v + 1) & 0xFF):
raise ValueError("Incorrect padding")
def import_openssh_private_generic(data, password):
# https://cvsweb.openbsd.org/cgi-bin/cvsweb/src/usr.bin/ssh/PROTOCOL.key?annotate=HEAD
# https://github.com/openssh/openssh-portable/blob/master/sshkey.c
# https://coolaj86.com/articles/the-openssh-private-key-format/
# https://coolaj86.com/articles/the-ssh-public-key-format/
if not data.startswith(b'openssh-key-v1\x00'):
raise ValueError("Incorrect magic value")
data = data[15:]
ciphername, data = read_string(data)
kdfname, data = read_string(data)
kdfoptions, data = read_bytes(data)
number_of_keys, data = read_int4(data)
if number_of_keys != 1:
raise ValueError("We only handle 1 key at a time")
_, data = read_string(data) # Public key
encrypted, data = read_bytes(data)
if data:
raise ValueError("Too much data")
if len(encrypted) % 8 != 0:
raise ValueError("Incorrect payload length")
# Decrypt if necessary
if ciphername == 'none':
decrypted = encrypted
else:
if (ciphername, kdfname) != ('aes256-ctr', 'bcrypt'):
raise ValueError("Unsupported encryption scheme %s/%s" % (ciphername, kdfname))
salt, kdfoptions = read_bytes(kdfoptions)
iterations, kdfoptions = read_int4(kdfoptions)
if len(salt) != 16:
raise ValueError("Incorrect salt length")
if kdfoptions:
raise ValueError("Too much data in kdfoptions")
pwd_sha512 = SHA512.new(password).digest()
# We need 32+16 = 48 bytes, therefore 2 bcrypt outputs are sufficient
stripes = []
constant = b"OxychromaticBlowfishSwatDynamite"
for count in range(1, 3):
salt_sha512 = SHA512.new(salt + struct.pack(">I", count)).digest()
out_le = _bcrypt_hash(pwd_sha512, 6, salt_sha512, constant, False)
out = struct.pack("<IIIIIIII", *struct.unpack(">IIIIIIII", out_le))
acc = bytearray(out)
for _ in range(1, iterations):
out_le = _bcrypt_hash(pwd_sha512, 6, SHA512.new(out).digest(), constant, False)
out = struct.pack("<IIIIIIII", *struct.unpack(">IIIIIIII", out_le))
strxor(acc, out, output=acc)
stripes.append(acc[:24])
result = b"".join([bchr(a)+bchr(b) for (a, b) in zip(*stripes)])
cipher = AES.new(result[:32],
AES.MODE_CTR,
nonce=b"",
initial_value=result[32:32+16])
decrypted = cipher.decrypt(encrypted)
checkint1, decrypted = read_int4(decrypted)
checkint2, decrypted = read_int4(decrypted)
if checkint1 != checkint2:
raise ValueError("Incorrect checksum")
ssh_name, decrypted = read_string(decrypted)
return ssh_name, decrypted

View File

@@ -0,0 +1,7 @@
from typing import Tuple
def read_int4(data: bytes) -> Tuple[int, bytes]: ...
def read_bytes(data: bytes) -> Tuple[bytes, bytes]: ...
def read_string(data: bytes) -> Tuple[str, bytes]: ...
def check_padding(pad: bytes) -> None: ...
def import_openssh_private_generic(data: bytes, password: bytes) -> Tuple[str, bytes]: ...

Binary file not shown.