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

pkcs1.py (5146B)


      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\rsa\pkcs1.py
      2 from __future__ import print_function
      3 import hashlib
      4 import os
      5 from rsa._compat import b
      6 from rsa import common, transform, core, varblock
      7 HASH_ASN1 = {'MD5': b('0 0\x0c\x06\x08*\x86H\x86\xf7\r\x02\x05\x05\x00\x04\x10'),
      8  'SHA-1': b('0!0\t\x06\x05+\x0e\x03\x02\x1a\x05\x00\x04\x14'),
      9  'SHA-256': b('010\r\x06\t`\x86H\x01e\x03\x04\x02\x01\x05\x00\x04 '),
     10  'SHA-384': b('0A0\r\x06\t`\x86H\x01e\x03\x04\x02\x02\x05\x00\x040'),
     11  'SHA-512': b('0Q0\r\x06\t`\x86H\x01e\x03\x04\x02\x03\x05\x00\x04@')}
     12 HASH_METHODS = {'MD5': hashlib.md5,
     13  'SHA-1': hashlib.sha1,
     14  'SHA-256': hashlib.sha256,
     15  'SHA-384': hashlib.sha384,
     16  'SHA-512': hashlib.sha512}
     17 
     18 class CryptoError(Exception):
     19     pass
     20 
     21 
     22 class DecryptionError(CryptoError):
     23     pass
     24 
     25 
     26 class VerificationError(CryptoError):
     27     pass
     28 
     29 
     30 def _pad_for_encryption(message, target_length):
     31     max_msglength = target_length - 11
     32     msglength = len(message)
     33     if msglength > max_msglength:
     34         raise OverflowError('%i bytes needed for message, but there is only space for %i' % (msglength, max_msglength))
     35     padding = b('')
     36     padding_length = target_length - msglength - 3
     37     while len(padding) < padding_length:
     38         needed_bytes = padding_length - len(padding)
     39         new_padding = os.urandom(needed_bytes + 5)
     40         new_padding = new_padding.replace(b('\x00'), b(''))
     41         padding = padding + new_padding[:needed_bytes]
     42 
     43     return b('').join([b('\x00\x02'),
     44      padding,
     45      b('\x00'),
     46      message])
     47 
     48 
     49 def _pad_for_signing(message, target_length):
     50     max_msglength = target_length - 11
     51     msglength = len(message)
     52     if msglength > max_msglength:
     53         raise OverflowError('%i bytes needed for message, but there is only space for %i' % (msglength, max_msglength))
     54     padding_length = target_length - msglength - 3
     55     return b('').join([b('\x00\x01'),
     56      padding_length * b('\xff'),
     57      b('\x00'),
     58      message])
     59 
     60 
     61 def encrypt(message, pub_key):
     62     keylength = common.byte_size(pub_key.n)
     63     padded = _pad_for_encryption(message, keylength)
     64     payload = transform.bytes2int(padded)
     65     encrypted = core.encrypt_int(payload, pub_key.e, pub_key.n)
     66     block = transform.int2bytes(encrypted, keylength)
     67     return block
     68 
     69 
     70 def decrypt(crypto, priv_key):
     71     blocksize = common.byte_size(priv_key.n)
     72     encrypted = transform.bytes2int(crypto)
     73     decrypted = core.decrypt_int(encrypted, priv_key.d, priv_key.n)
     74     cleartext = transform.int2bytes(decrypted, blocksize)
     75     if cleartext[0:2] != b('\x00\x02'):
     76         raise DecryptionError('Decryption failed')
     77     try:
     78         sep_idx = cleartext.index(b('\x00'), 2)
     79     except ValueError:
     80         raise DecryptionError('Decryption failed')
     81 
     82     return cleartext[sep_idx + 1:]
     83 
     84 
     85 def sign(message, priv_key, hash):
     86     if hash not in HASH_ASN1:
     87         raise ValueError('Invalid hash method: %s' % hash)
     88     asn1code = HASH_ASN1[hash]
     89     hash = _hash(message, hash)
     90     cleartext = asn1code + hash
     91     keylength = common.byte_size(priv_key.n)
     92     padded = _pad_for_signing(cleartext, keylength)
     93     payload = transform.bytes2int(padded)
     94     encrypted = core.encrypt_int(payload, priv_key.d, priv_key.n)
     95     block = transform.int2bytes(encrypted, keylength)
     96     return block
     97 
     98 
     99 def verify(message, signature, pub_key):
    100     blocksize = common.byte_size(pub_key.n)
    101     encrypted = transform.bytes2int(signature)
    102     decrypted = core.decrypt_int(encrypted, pub_key.e, pub_key.n)
    103     clearsig = transform.int2bytes(decrypted, blocksize)
    104     if clearsig[0:2] != b('\x00\x01'):
    105         raise VerificationError('Verification failed')
    106     try:
    107         sep_idx = clearsig.index(b('\x00'), 2)
    108     except ValueError:
    109         raise VerificationError('Verification failed')
    110 
    111     method_name, signature_hash = _find_method_hash(clearsig[sep_idx + 1:])
    112     message_hash = _hash(message, method_name)
    113     if message_hash != signature_hash:
    114         raise VerificationError('Verification failed')
    115     return True
    116 
    117 
    118 def _hash(message, method_name):
    119     if method_name not in HASH_METHODS:
    120         raise ValueError('Invalid hash method: %s' % method_name)
    121     method = HASH_METHODS[method_name]
    122     hasher = method()
    123     if hasattr(message, 'read') and hasattr(message.read, '__call__'):
    124         for block in varblock.yield_fixedblocks(message, 1024):
    125             hasher.update(block)
    126 
    127     else:
    128         hasher.update(message)
    129     return hasher.digest()
    130 
    131 
    132 def _find_method_hash(method_hash):
    133     for hashname, asn1code in HASH_ASN1.items():
    134         if not method_hash.startswith(asn1code):
    135             continue
    136         return (hashname, method_hash[len(asn1code):])
    137 
    138     raise VerificationError('Verification failed')
    139 
    140 
    141 __all__ = ['encrypt',
    142  'decrypt',
    143  'sign',
    144  'verify',
    145  'DecryptionError',
    146  'VerificationError',
    147  'CryptoError']
    148 if __name__ == '__main__':
    149     print('Running doctests 1000x or until failure')
    150     import doctest
    151     for count in range(1000):
    152         failures, tests = doctest.testmod()
    153         if failures:
    154             break
    155         if count and count % 100 == 0:
    156             print('%i times' % count)
    157 
    158     print('Doctests done')