Added PKCS#1 signatures and verification of signatures

3 files changed, 224 insertions, 18 deletions

diff --git a/rsa/pkcs1.py b/rsa/pkcs1.py
index d84c24c..3cca453 100644
--- a/rsa/pkcs1.py
+++ b/rsa/pkcs1.py
@@ -6,13 +6,43 @@ very clear example, read http://www.di-mgt.com.au/rsa_alg.html#pkcs1schemes
 At least 8 bytes of random padding is used when encrypting a message. This makes
 these methods much more secure than the ones in the ``rsa`` module.
 
+WARNING: this module leaks information when decryption or verification fails.
+The exceptions that are raised contain the Python traceback information, which
+can be used to deduce where in the process the failure occurred. DO NOT PASS
+SUCH INFORMATION to your users.
 '''
 
+import hashlib
 import os
 
 from rsa import common, transform, core
 
+# ASN.1 codes that describe the hash algorithm used.
+HASH_ASN1 = {
+    'MD5': '\x30\x20\x30\x0c\x06\x08\x2a\x86\x48\x86\xf7\x0d\x02\x05\x05\x00\x04\x10',
+    'SHA-1': '\x30\x21\x30\x09\x06\x05\x2b\x0e\x03\x02\x1a\x05\x00\x04\x14',
+    'SHA-256': '\x30\x31\x30\x0d\x06\x09\x60\x86\x48\x01\x65\x03\x04\x02\x01\x05\x00\x04\x20',
+    'SHA-384': '\x30\x41\x30\x0d\x06\x09\x60\x86\x48\x01\x65\x03\x04\x02\x02\x05\x00\x04\x30',
+    'SHA-512': '\x30\x51\x30\x0d\x06\x09\x60\x86\x48\x01\x65\x03\x04\x02\x03\x05\x00\x04\x40',
+}
 
+HASH_METHODS = {
+    'MD5': hashlib.md5,
+    'SHA-1': hashlib.sha1,
+    'SHA-256': hashlib.sha256,
+    'SHA-384': hashlib.sha384,
+    'SHA-512': hashlib.sha512,
+}
+
+class CryptoError(Exception):
+    '''Base class for all exceptions in this module.'''
+
+class DecryptionError(CryptoError):
+    '''Raised when decryption fails.'''
+
+class VerificationError(CryptoError):
+    '''Raised when verification fails.'''
+ 
 def _pad_for_encryption(message, target_length):
     r'''Pads the message for encryption, returning the padded message.
     
@@ -58,6 +88,40 @@ def _pad_for_encryption(message, target_length):
                     '\x00',
                     message])
     
+
+def _pad_for_signing(message, target_length):
+    r'''Pads the message for signing, returning the padded message.
+    
+    The padding is always a repetition of FF bytes.
+    
+    @return: 00 01 PADDING 00 MESSAGE
+    
+    >>> block = _pad_for_signing('hello', 16)
+    >>> len(block)
+    16
+    >>> block[0:2]
+    '\x00\x01'
+    >>> block[-6:]
+    '\x00hello'
+    >>> block[2:-6]
+    '\xff\xff\xff\xff\xff\xff\xff\xff'
+    
+    '''
+
+    max_msglength = target_length - 11
+    msglength = len(message)
+    
+    if msglength > max_msglength:
+        raise OverflowError('%i bytes needed for message, but there is only'
+            ' space for %i' % (msglength, max_msglength))
+    
+    padding_length = target_length - msglength - 3
+    
+    return ''.join(['\x00\x01',
+                    padding_length * '\xff',
+                    '\x00',
+                    message])
+    
     
 def encrypt(message, pub_key):
     '''Encrypts the given message using PKCS1 v1.5
@@ -100,9 +164,9 @@ def decrypt(crypto, priv_key):
     @param crypto: the crypto text as returned by ``encrypt(message, pub_key)``
     @param priv_key: the private key to decrypt with.
     
-    @raise ValueError: when the decryption fails. No details are given as to why
-        the code thinks the decryption fails, as this would leak information
-        about the private key.
+    @raise DecryptionError: when the decryption fails. No details are given as
+        to why the code thinks the decryption fails, as this would leak
+        information about the private key.
 
     >>> from rsa import keygen, common
     >>> (pub_key, priv_key) = keygen.newkeys(256)
@@ -124,16 +188,123 @@ def decrypt(crypto, priv_key):
 
     # If we can't find the cleartext marker, decryption failed.
     if cleartext[0:2] != '\x00\x02':
-        raise ValueError('Decryption failed')
+        raise DecryptionError('Decryption failed')
     
     # Find the 00 separator between the padding and the message
     try:
         sep_idx = cleartext.index('\x00', 2)
     except ValueError:
-        raise ValueError('Decryption failed')
+        raise DecryptionError('Decryption failed')
     
     return cleartext[sep_idx+1:]
     
+def sign(message, priv_key, hash):
+    '''Signs the message with the private key.
+
+    Hashes the message, then signs the hash with the given key. This is known
+    as a "detached signature", because the message itself isn't signed.
+    
+    @param message: the message to sign
+    @param priv_key: the private key to sign with
+    @param hash: the hash method used on the message. Use 'MD5', 'SHA-1',
+        'SHA-256', 'SHA-384' or 'SHA-512'.
+    
+    @return: a message signature block.
+    
+    @raise OverflowError: if the private key is too small to contain the
+        requested hash.
+
+    '''
+
+    # Get the ASN1 code for this hash method
+    if hash not in HASH_ASN1:
+        raise ValueError('Invalid hash method: %s' % hash)
+    asn1code = HASH_ASN1[hash]
+    
+    # Calculate the hash
+    hash = _hash(message, hash)
+
+    # Encrypt the hash with the private key
+    cleartext = asn1code + hash
+    keylength = common.byte_size(priv_key['n'])
+    padded = _pad_for_signing(cleartext, keylength)
+    
+    payload = transform.bytes2int(padded)
+    encrypted = core.encrypt_int(payload, priv_key['d'], priv_key['n'])
+    block = transform.int2bytes(encrypted, keylength)
+    
+    return block
+
+def verify(message, signature, pub_key):
+    '''Verifies that the signature matches the message.
+    
+    The hash method is detected automatically from the signature.
+    
+    @param message: the signed message
+    @param signature: the signature block, as created with ``sign(...)``.
+    @param pub_key: the public key of the person signing the message.
+    
+    @raise VerificationError: when the signature doesn't match the message.
+    '''
+    
+    blocksize = common.byte_size(pub_key['n']) 
+    encrypted = transform.bytes2int(signature)
+    decrypted = core.decrypt_int(encrypted, pub_key['e'], pub_key['n'])
+    clearsig = transform.int2bytes(decrypted, blocksize)
+
+    # If we can't find the signature  marker, verification failed.
+    if clearsig[0:2] != '\x00\x01':
+        raise VerificationError('Verification failed')
+    
+    # Find the 00 separator between the padding and the payload
+    try:
+        sep_idx = clearsig.index('\x00', 2)
+    except ValueError:
+        raise VerificationError('Verification failed')
+    
+    # Get the hash and the hash method
+    (method_name, signature_hash) = _find_method_hash(clearsig[sep_idx+1:])
+    message_hash = _hash(message, method_name)
+
+    # Compare the real hash to the hash in the signature
+    if message_hash != signature_hash:
+        raise VerificationError('Verification failed')
+
+def _hash(message, method_name):
+    '''Returns the message digest.'''
+
+    if method_name not in HASH_METHODS:
+        raise ValueError('Invalid hash method: %s' % method_name)
+    
+    method = HASH_METHODS[method_name]
+    hasher = method()
+    hasher.update(message)
+    return hasher.digest()
+
+
+def _find_method_hash(method_hash):
+    '''Finds the hash method and the hash itself.
+    
+    @param method_hash: ASN1 code for the hash method concatenated with the
+        hash itself.
+    
+    @return: tuple (method, hash) where ``method`` is the used hash method, and
+        ``hash`` is the hash itself.
+    
+    @raise VerificationFailed: when the hash method cannot be found
+    '''
+
+    for (hashname, asn1code) in HASH_ASN1.iteritems():
+        if not method_hash.startswith(asn1code):
+            continue
+        
+        return (hashname, method_hash[len(asn1code):])
+    
+    raise VerificationError('Verification failed')
+
+
+__all__ = ['encrypt', 'decript', 'sign', 'verify',
+           'DecryptionError', 'VerificationError', 'CryptoError']
 
 if __name__ == '__main__':
     print 'Running doctests 1000x or until failure'
diff --git a/rsa/transform.py b/rsa/transform.py
index 3f151ad..608d838 100755
--- a/rsa/transform.py
+++ b/rsa/transform.py
@@ -66,6 +66,9 @@ def int2bytes(number, block_size=None):
         raise TypeError("You must pass an integer for 'number', not %s" %
             number.__class__)
 
+    if number < 0:
+        raise ValueError('Negative numbers cannot be used: %i' % number)
+
     # Do some bounds checking
     if block_size is not None:
         needed_bytes = common.byte_size(number)
diff --git a/tests/test_pkcs1.py b/tests/test_pkcs1.py
index 0d55d7b..3392ed7 100644
--- a/tests/test_pkcs1.py
+++ b/tests/test_pkcs1.py
@@ -32,7 +32,8 @@ class BinaryTest(unittest.TestCase):
         # Alter the encrypted stream
         encrypted = encrypted[:5] + chr(ord(encrypted[5]) + 1) + encrypted[6:]
         
-        self.assertRaises(ValueError, pkcs1.decrypt, encrypted, self.priv)
+        self.assertRaises(pkcs1.DecryptionError, pkcs1.decrypt, encrypted,
+                          self.priv)
 
     def test_randomness(self):
         '''Encrypting the same message twice should result in different
@@ -45,15 +46,46 @@ class BinaryTest(unittest.TestCase):
         
         self.assertNotEqual(encrypted1, encrypted2)
 
-#    def test_sign_verify(self):
-#
-#        message = struct.pack('>IIII', 0, 0, 0, 1) + 20 * '\x00'
-#        print "\tMessage:   %r" % message
-#
-#        signed = rsa.sign(message, self.priv)
-#        print "\tSigned:    %r" % signed
-#
-#        verified = rsa.verify(signed, self.pub)
-#        print "\tVerified:  %r" % verified
-#
-#        self.assertEqual(message, verified)
+class SignatureTest(unittest.TestCase):
+
+    def setUp(self):
+        (self.pub, self.priv) = rsa.newkeys(512)
+
+    def test_sign_verify(self):
+        '''Test happy flow of sign and verify'''
+        
+        message = 'je moeder'
+        print "\tMessage:   %r" % message
+
+        signature = pkcs1.sign(message, self.priv, 'SHA-256')
+        print "\tSignature: %r" % signature
+
+        pkcs1.verify(message, signature, self.pub)
+
+    def test_alter_message(self):
+        '''Altering the message should let the verification fail.'''
+        
+        signature = pkcs1.sign('je moeder', self.priv, 'SHA-256')
+        self.assertRaises(pkcs1.VerificationError, pkcs1.verify,
+                          'mijn moeder', signature, self.pub)
+
+    def test_sign_different_key(self):
+        '''Signing with another key should let the verification fail.'''
+        
+        (otherpub, _) = rsa.newkeys(512)
+        
+        message = 'je moeder'
+        signature = pkcs1.sign(message, self.priv, 'SHA-256')
+        self.assertRaises(pkcs1.VerificationError, pkcs1.verify,
+                          message, signature, otherpub)
+
+    def test_multiple_signings(self):
+        '''Signing the same message twice should return the same signatures.'''
+        
+        message = struct.pack('>IIII', 0, 0, 0, 1)
+        signature1 = pkcs1.sign(message, self.priv, 'SHA-1')
+        signature2 = pkcs1.sign(message, self.priv, 'SHA-1')
+        
+        self.assertEqual(signature1, signature2)
+
+        
\ No newline at end of file