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')