NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

key.py (8498B)


      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\rsa\key.py
      2 from __future__ import print_function
      3 import logging
      4 from rsa._compat import b, bytes_type
      5 import rsa.prime
      6 import rsa.pem
      7 import rsa.common
      8 log = logging.getLogger(__name__)
      9 
     10 class AbstractKey(object):
     11 
     12     @classmethod
     13     def load_pkcs1(cls, keyfile, format = 'PEM'):
     14         methods = {'PEM': cls._load_pkcs1_pem,
     15          'DER': cls._load_pkcs1_der}
     16         if format not in methods:
     17             formats = ', '.join(sorted(methods.keys()))
     18             raise ValueError('Unsupported format: %r, try one of %s' % (format, formats))
     19         method = methods[format]
     20         return method(keyfile)
     21 
     22     def save_pkcs1(self, format = 'PEM'):
     23         methods = {'PEM': self._save_pkcs1_pem,
     24          'DER': self._save_pkcs1_der}
     25         if format not in methods:
     26             formats = ', '.join(sorted(methods.keys()))
     27             raise ValueError('Unsupported format: %r, try one of %s' % (format, formats))
     28         method = methods[format]
     29         return method()
     30 
     31 
     32 class PublicKey(AbstractKey):
     33     __slots__ = ('n', 'e')
     34 
     35     def __init__(self, n, e):
     36         self.n = n
     37         self.e = e
     38 
     39     def __getitem__(self, key):
     40         return getattr(self, key)
     41 
     42     def __repr__(self):
     43         return 'PublicKey(%i, %i)' % (self.n, self.e)
     44 
     45     def __eq__(self, other):
     46         if other is None:
     47             return False
     48         if not isinstance(other, PublicKey):
     49             return False
     50         return self.n == other.n and self.e == other.e
     51 
     52     def __ne__(self, other):
     53         return not self == other
     54 
     55     @classmethod
     56     def _load_pkcs1_der(cls, keyfile):
     57         from pyasn1.codec.der import decoder
     58         from rsa.asn1 import AsnPubKey
     59         priv, _ = decoder.decode(keyfile, asn1Spec=AsnPubKey())
     60         return cls(n=int(priv['modulus']), e=int(priv['publicExponent']))
     61 
     62     def _save_pkcs1_der(self):
     63         from pyasn1.codec.der import encoder
     64         from rsa.asn1 import AsnPubKey
     65         asn_key = AsnPubKey()
     66         asn_key.setComponentByName('modulus', self.n)
     67         asn_key.setComponentByName('publicExponent', self.e)
     68         return encoder.encode(asn_key)
     69 
     70     @classmethod
     71     def _load_pkcs1_pem(cls, keyfile):
     72         der = rsa.pem.load_pem(keyfile, 'RSA PUBLIC KEY')
     73         return cls._load_pkcs1_der(der)
     74 
     75     def _save_pkcs1_pem(self):
     76         der = self._save_pkcs1_der()
     77         return rsa.pem.save_pem(der, 'RSA PUBLIC KEY')
     78 
     79     @classmethod
     80     def load_pkcs1_openssl_pem(cls, keyfile):
     81         der = rsa.pem.load_pem(keyfile, 'PUBLIC KEY')
     82         return cls.load_pkcs1_openssl_der(der)
     83 
     84     @classmethod
     85     def load_pkcs1_openssl_der(cls, keyfile):
     86         from rsa.asn1 import OpenSSLPubKey
     87         from pyasn1.codec.der import decoder
     88         from pyasn1.type import univ
     89         keyinfo, _ = decoder.decode(keyfile, asn1Spec=OpenSSLPubKey())
     90         if keyinfo['header']['oid'] != univ.ObjectIdentifier('1.2.840.113549.1.1.1'):
     91             raise TypeError('This is not a DER-encoded OpenSSL-compatible public key')
     92         return cls._load_pkcs1_der(keyinfo['key'][1:])
     93 
     94 
     95 class PrivateKey(AbstractKey):
     96     __slots__ = ('n', 'e', 'd', 'p', 'q', 'exp1', 'exp2', 'coef')
     97 
     98     def __init__(self, n, e, d, p, q, exp1 = None, exp2 = None, coef = None):
     99         self.n = n
    100         self.e = e
    101         self.d = d
    102         self.p = p
    103         self.q = q
    104         if exp1 is None:
    105             self.exp1 = int(d % (p - 1))
    106         else:
    107             self.exp1 = exp1
    108         if exp1 is None:
    109             self.exp2 = int(d % (q - 1))
    110         else:
    111             self.exp2 = exp2
    112         if coef is None:
    113             self.coef = rsa.common.inverse(q, p)
    114         else:
    115             self.coef = coef
    116 
    117     def __getitem__(self, key):
    118         return getattr(self, key)
    119 
    120     def __repr__(self):
    121         return 'PrivateKey(%(n)i, %(e)i, %(d)i, %(p)i, %(q)i)' % self
    122 
    123     def __eq__(self, other):
    124         if other is None:
    125             return False
    126         if not isinstance(other, PrivateKey):
    127             return False
    128         return self.n == other.n and self.e == other.e and self.d == other.d and self.p == other.p and self.q == other.q and self.exp1 == other.exp1 and self.exp2 == other.exp2 and self.coef == other.coef
    129 
    130     def __ne__(self, other):
    131         return not self == other
    132 
    133     @classmethod
    134     def _load_pkcs1_der(cls, keyfile):
    135         from pyasn1.codec.der import decoder
    136         priv, _ = decoder.decode(keyfile)
    137         if priv[0] != 0:
    138             raise ValueError('Unable to read this file, version %s != 0' % priv[0])
    139         as_ints = tuple((int(x) for x in priv[1:9]))
    140         return cls(*as_ints)
    141 
    142     def _save_pkcs1_der(self):
    143         from pyasn1.type import univ, namedtype
    144         from pyasn1.codec.der import encoder
    145 
    146         class AsnPrivKey(univ.Sequence):
    147             componentType = namedtype.NamedTypes(namedtype.NamedType('version', univ.Integer()), namedtype.NamedType('modulus', univ.Integer()), namedtype.NamedType('publicExponent', univ.Integer()), namedtype.NamedType('privateExponent', univ.Integer()), namedtype.NamedType('prime1', univ.Integer()), namedtype.NamedType('prime2', univ.Integer()), namedtype.NamedType('exponent1', univ.Integer()), namedtype.NamedType('exponent2', univ.Integer()), namedtype.NamedType('coefficient', univ.Integer()))
    148 
    149         asn_key = AsnPrivKey()
    150         asn_key.setComponentByName('version', 0)
    151         asn_key.setComponentByName('modulus', self.n)
    152         asn_key.setComponentByName('publicExponent', self.e)
    153         asn_key.setComponentByName('privateExponent', self.d)
    154         asn_key.setComponentByName('prime1', self.p)
    155         asn_key.setComponentByName('prime2', self.q)
    156         asn_key.setComponentByName('exponent1', self.exp1)
    157         asn_key.setComponentByName('exponent2', self.exp2)
    158         asn_key.setComponentByName('coefficient', self.coef)
    159         return encoder.encode(asn_key)
    160 
    161     @classmethod
    162     def _load_pkcs1_pem(cls, keyfile):
    163         der = rsa.pem.load_pem(keyfile, b('RSA PRIVATE KEY'))
    164         return cls._load_pkcs1_der(der)
    165 
    166     def _save_pkcs1_pem(self):
    167         der = self._save_pkcs1_der()
    168         return rsa.pem.save_pem(der, b('RSA PRIVATE KEY'))
    169 
    170 
    171 def find_p_q(nbits, getprime_func = rsa.prime.getprime, accurate = True):
    172     total_bits = nbits * 2
    173     shift = nbits // 16
    174     pbits = nbits + shift
    175     qbits = nbits - shift
    176     log.debug('find_p_q(%i): Finding p', nbits)
    177     p = getprime_func(pbits)
    178     log.debug('find_p_q(%i): Finding q', nbits)
    179     q = getprime_func(qbits)
    180 
    181     def is_acceptable(p, q):
    182         if p == q:
    183             return False
    184         if not accurate:
    185             return True
    186         found_size = rsa.common.bit_size(p * q)
    187         return total_bits == found_size
    188 
    189     change_p = False
    190     while not is_acceptable(p, q):
    191         if change_p:
    192             p = getprime_func(pbits)
    193         else:
    194             q = getprime_func(qbits)
    195         change_p = not change_p
    196 
    197     return (max(p, q), min(p, q))
    198 
    199 
    200 def calculate_keys(p, q, nbits):
    201     phi_n = (p - 1) * (q - 1)
    202     e = 65537
    203     try:
    204         d = rsa.common.inverse(e, phi_n)
    205     except ValueError:
    206         raise ValueError('e (%d) and phi_n (%d) are not relatively prime' % (e, phi_n))
    207 
    208     if e * d % phi_n != 1:
    209         raise ValueError('e (%d) and d (%d) are not mult. inv. modulo phi_n (%d)' % (e, d, phi_n))
    210     return (e, d)
    211 
    212 
    213 def gen_keys(nbits, getprime_func, accurate = True):
    214     p, q = find_p_q(nbits // 2, getprime_func, accurate)
    215     e, d = calculate_keys(p, q, nbits // 2)
    216     return (p,
    217      q,
    218      e,
    219      d)
    220 
    221 
    222 def newkeys(nbits, accurate = True, poolsize = 1):
    223     if nbits < 16:
    224         raise ValueError('Key too small')
    225     if poolsize < 1:
    226         raise ValueError('Pool size (%i) should be >= 1' % poolsize)
    227     if poolsize > 1:
    228         from rsa import parallel
    229         import functools
    230         getprime_func = functools.partial(parallel.getprime, poolsize=poolsize)
    231     else:
    232         getprime_func = rsa.prime.getprime
    233     p, q, e, d = gen_keys(nbits, getprime_func)
    234     n = p * q
    235     return (PublicKey(n, e), PrivateKey(n, e, d, p, q))
    236 
    237 
    238 __all__ = ['PublicKey', 'PrivateKey', 'newkeys']
    239 if __name__ == '__main__':
    240     import doctest
    241     try:
    242         for count in range(100):
    243             failures, tests = doctest.testmod()
    244             if failures:
    245                 break
    246             if count and count % 10 == 0 or count == 1:
    247                 print('%i times' % count)
    248 
    249     except KeyboardInterrupt:
    250         print('Aborted')
    251     else:
    252         print('Doctests done')