Added PKCS#1 v1.5 encryption

4 files changed, 312 insertions, 1 deletions

diff --git a/lib/Crypto/Cipher/PKCS1_v1_5.py b/lib/Crypto/Cipher/PKCS1_v1_5.py
new file mode 100644
index 0000000..b6477c4
--- /dev/null
+++ b/lib/Crypto/Cipher/PKCS1_v1_5.py
@@ -0,0 +1,142 @@
+# -*- coding: utf-8 -*-
+#
+#  Cipher/PKCS1-v1_5.py : PKCS#1 v1.5
+#
+# ===================================================================
+# 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.
+# ===================================================================
+
+"""RSA encryption protocol according to PKCS#1 v1.5
+
+See RFC3447 or the original RSA Labs specification at
+http://www.rsa.com/rsalabs/node.asp?id=2125.
+
+This scheme is more properly called ``RSAES-PKCS1-v1_5``.
+
+As an example, a sender may encrypt a message in this way:
+
+        >>> from Crypto.Cipher import PKCS1_v1_5
+        >>> from Crypto.PublicKey import RSA
+        >>> from Crypto import Random
+        >>>
+        >>> message = 'To be encrypted'
+        >>> key = RSA.importKey('pubkey.der')
+        >>> rng = Random.new().read
+        >>> ciphertext = PKCS1_v1_5.encrypt(message, key, rng)
+
+At the receiver side, decryption can be done using the private part of
+the RSA key:
+
+        >>> key = RSA.importKey('privkey.der')
+        >>> message = PKCS1_v1_5.decrypt(ciphertext):
+"""
+
+__revision__ = "$Id$"
+__all__ = [ 'encrypt', 'decrypt' ]
+
+from Crypto.Util.number import ceil_div
+import Crypto.Util.number
+
+def encrypt(message, key, randFunc):
+    """Produce the PKCS#1 v1.5 encryption of a message.
+
+    This function is named ``RSAES-PKCS1-V1_5-ENCRYPT``, and is specified in
+    section 7.2.1 of RFC3447.
+
+    :Parameters:
+     message : string
+            The message to encrypt, also known as plaintext. It can be of
+            variable length, but not longer than the RSA modulus (in bytes) minus 11.
+     key : RSA key object
+            The key to use to encrypt the message. This is a `Crypto.PublicKey.RSA`
+            object.
+     randFunc : callable
+            An RNG function that accepts as only parameter an integer, and returns
+            a string of random bytes.
+
+    :Return: A string, the ciphertext in which the message is encrypted.
+        It is as long as the RSA modulus (in bytes).
+    :Raise ValueError:
+        If the RSA key length is not sufficiently long to deal with the given
+        message.
+    """
+    # TODO: Verify the key is RSA
+
+    # See 7.2.1 in RFC3447
+    modBits = Crypto.Util.number.size(key.n)
+    k = ceil_div(modBits,8) # Convert from bits to bytes
+    mLen = len(message)
+
+    # Step 1
+    if mLen > k-11:
+        raise ValueError("Plaintext is too long.")
+    # Step 2a
+    class nonZeroRandByte:
+        def __init__(self, rf): self.rf=rf
+        def __call__(self, c):
+            while c=='\x00': c=self.rf(1)
+            return c
+    ps = "".join(map(nonZeroRandByte(randFunc), randFunc(k-mLen-3)))
+    # Step 2b
+    em = '\x00\x02' + ps + '\x00' + message
+    # Step 3a (OS2IP), step 3b (RSAEP), part of step 3c (I2OSP)
+    m = key.encrypt(em, 0)[0]
+    # Complete step 3c (I2OSP)
+    c = '\x00'*(k-len(m)) + m
+    return c
+
+def decrypt(ct, key):
+    """Decrypt a PKCS#1 v1.5 ciphertext.
+
+    This function is named ``RSAES-PKCS1-V1_5-DECRYPT``, and is specified in
+    section 7.2.2 of RFC3447.
+
+    :Parameters:
+     ct : string
+            The ciphertext that contains the message to recover.
+     key : RSA key object
+            The key to use to verify the message. This is a `Crypto.PublicKey.RSA`
+            object. It must have its private half.
+
+    :Return: A string, the original message.
+    :Raise ValueError:
+        If the ciphertext length is incorrect, or if the encryption does not
+        succeed.
+    :Raise TypeError:
+        If the RSA key has no private half.
+    """
+    # TODO: Verify the key is RSA
+
+    # See 7.2.1 in RFC3447
+    modBits = Crypto.Util.number.size(key.n)
+    k = ceil_div(modBits,8) # Convert from bits to bytes
+
+    # Step 1
+    if len(ct) != k:
+        raise ValueError("Ciphertext with incorrect length.")
+    # Step 2a (O2SIP), 2b (RSADP), and part of 2c (I2OSP)
+    m = key.decrypt(ct)
+    # Complete step 2c (I2OSP)
+    em = '\x00'*(k-len(m)) + m
+    # Step 3
+    sep = em.find('\x00',2)
+    #print "sep=", sep
+    if  not em.startswith('\x00\x02') or sep<10:
+        raise ValueError("Incorrect decryption.")
+    # Step 4
+    return em[sep+1:]
+
diff --git a/lib/Crypto/Cipher/__init__.py b/lib/Crypto/Cipher/__init__.py
index d8ceed9..6b47771 100644
--- a/lib/Crypto/Cipher/__init__.py
+++ b/lib/Crypto/Cipher/__init__.py
@@ -43,7 +43,8 @@ Crypto.Cipher.XOR         The simple XOR cipher.
 
 __all__ = ['AES', 'ARC2', 'ARC4',
            'Blowfish', 'CAST', 'DES', 'DES3',
-           'XOR'
+           'XOR',
+           'PKCS1_v1_5'
            ]
 
 __revision__ = "$Id$"
diff --git a/lib/Crypto/SelfTest/Cipher/__init__.py b/lib/Crypto/SelfTest/Cipher/__init__.py
index 0034f37..411bf81 100644
--- a/lib/Crypto/SelfTest/Cipher/__init__.py
+++ b/lib/Crypto/SelfTest/Cipher/__init__.py
@@ -36,6 +36,7 @@ def get_tests(config={}):
     import test_DES3;     tests += test_DES3.get_tests(config=config)
     import test_DES;      tests += test_DES.get_tests(config=config)
     import test_XOR;      tests += test_XOR.get_tests(config=config)
+    import test_pkcs_15;  tests += test_pkcs1_15.get_tests(config=config)
     return tests
 
 if __name__ == '__main__':
diff --git a/lib/Crypto/SelfTest/Cipher/test_pkcs1_15.py b/lib/Crypto/SelfTest/Cipher/test_pkcs1_15.py
new file mode 100644
index 0000000..8706b30
--- /dev/null
+++ b/lib/Crypto/SelfTest/Cipher/test_pkcs1_15.py
@@ -0,0 +1,167 @@
+# -*- coding: utf-8 -*-
+#
+#  SelfTest/Cipher/test_pkcs1_15.py: Self-test for PKCS#1 v1.5 encryption
+#
+# ===================================================================
+# 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.
+# ===================================================================
+
+__revision__ = "$Id$"
+
+import unittest
+
+from string import maketrans
+
+from Crypto.PublicKey import RSA
+from Crypto.SelfTest.st_common import list_test_cases, a2b_hex, b2a_hex
+from Crypto import Random
+from Crypto.Cipher import PKCS1_v1_5 as PKCS
+
+def rws(t):
+    """Remove white spaces, tabs, and new lines from a string"""
+    return t.translate(maketrans("",""),'\n\t ')
+
+def t2b(t):
+    """Convert a text string with bytes in hex form to a byte string"""
+    clean = rws(t)
+    if len(clean)%2 == 1:
+        raise ValueError("Even number of characters expected")
+    return a2b_hex(clean)
+
+class PKCS1_15_Tests(unittest.TestCase):
+
+        def setUp(self):
+                self.rng = Random.new().read
+                self.key1024 = RSA.generate(1024, self.rng)
+
+        # List of tuples with test data for PKCS#1 v1.5.
+        # Each tuple is made up by:
+        #       Item #0: dictionary with RSA key component, or key to import
+        #       Item #1: plaintext
+        #       Item #2: ciphertext
+        #       Item #3: random data
+
+        _testData = (
+
+                #
+                # Generated with openssl 0.9.8o
+                #
+                (
+                # Private key
+                '''-----BEGIN RSA PRIVATE KEY-----
+MIICXAIBAAKBgQDAiAnvIAOvqVwJTaYzsKnefZftgtXGE2hPJppGsWl78yz9jeXY
+W/FxX/gTPURArNhdnhP6n3p2ZaDIBrO2zizbgIXs0IsljTTcr4vnI8fMXzyNUOjA
+zP3nzMqZDZK6757XQAobOssMkBFqRWwilT/3DsBhRpl3iMUhF+wvpTSHewIDAQAB
+AoGAC4HV/inOrpgTvSab8Wj0riyZgQOZ3U3ZpSlsfR8ra9Ib9Uee3jCYnKscu6Gk
+y6zI/cdt8EPJ4PuwAWSNJzbpbVaDvUq25OD+CX8/uRT08yBS4J8TzBitZJTD4lS7
+atdTnKT0Wmwk+u8tDbhvMKwnUHdJLcuIsycts9rwJVapUtkCQQDvDpx2JMun0YKG
+uUttjmL8oJ3U0m3ZvMdVwBecA0eebZb1l2J5PvI3EJD97eKe91Nsw8T3lwpoN40k
+IocSVDklAkEAzi1HLHE6EzVPOe5+Y0kGvrIYRRhncOb72vCvBZvD6wLZpQgqo6c4
+d3XHFBBQWA6xcvQb5w+VVEJZzw64y25sHwJBAMYReRl6SzL0qA0wIYrYWrOt8JeQ
+8mthulcWHXmqTgC6FEXP9Es5GD7/fuKl4wqLKZgIbH4nqvvGay7xXLCXD/ECQH9a
+1JYNMtRen5unSAbIOxRcKkWz92F0LKpm9ZW/S9vFHO+mBcClMGoKJHiuQxLBsLbT
+NtEZfSJZAeS2sUtn3/0CQDb2M2zNBTF8LlM0nxmh0k9VGm5TVIyBEMcipmvOgqIs
+HKukWBcq9f/UOmS0oEhai/6g+Uf7VHJdWaeO5LzuvwU=
+-----END RSA PRIVATE KEY-----''',
+                # Plaintext
+                '''THIS IS PLAINTEXT\x0A''',
+                # Ciphertext
+                '''3f dc fd 3c cd 5c 9b 12  af 65 32 e3 f7 d0 da 36
+                8f 8f d9 e3 13 1c 7f c8  b3 f9 c1 08 e4 eb 79 9c
+                91 89 1f 96 3b 94 77 61  99 a4 b1 ee 5d e6 17 c9
+                5d 0a b5 63 52 0a eb 00  45 38 2a fb b0 71 3d 11
+                f7 a1 9e a7 69 b3 af 61  c0 bb 04 5b 5d 4b 27 44
+                1f 5b 97 89 ba 6a 08 95  ee 4f a2 eb 56 64 e5 0f
+                da 7c f9 9a 61 61 06 62  ed a0 bc 5f aa 6c 31 78
+                70 28 1a bb 98 3c e3 6a  60 3c d1 0b 0f 5a f4 75''',
+                # Random data
+                '''eb d7 7d 86 a4 35 23 a3 54 7e 02 0b 42 1d
+                61 6c af 67 b8 4e 17 56 80 66 36 04 64 34 26 8a
+                47 dd 44 b3 1a b2 17 60 f4 91 2e e2 b5 95 64 cc
+                f9 da c8 70 94 54 86 4c ef 5b 08 7d 18 c4 ab 8d
+                04 06 33 8f ca 15 5f 52 60 8a a1 0c f5 08 b5 4c
+                bb 99 b8 94 25 04 9c e6 01 75 e6 f9 63 7a 65 61
+                13 8a a7 47 77 81 ae 0d b8 2c 4d 50 a5'''
+                ),
+        )
+
+        def testEncrypt1(self):
+                for test in self._testData:
+                        # Build the key
+                        key = RSA.importKey(test[0])
+                        # RNG that takes its random numbers from a pool given
+                        # at initialization
+                        class randGen:
+                            def __init__(self, data):
+                                self.data = data
+                                self.idx = 0
+                            def __call__(self, N):
+                                r = self.data[self.idx:N]
+                                self.idx += N
+                                return r
+                        # The real test
+                        ct = PKCS.encrypt(test[1], key, randGen(t2b(test[3])))
+                        self.assertEqual(ct, t2b(test[2]))
+
+        def testEncrypt2(self):
+                # Verify that encryption fail if plaintext is too long
+                pt = '\x00'*(128-11+1)
+                self.assertRaises(ValueError, PKCS.encrypt, pt, self.key1024, self.rng)
+
+        def testVerify1(self):
+                for test in self._testData:
+                        # Build the key
+                        key = RSA.importKey(test[0])
+                        # The real test
+                        pt = PKCS.decrypt(t2b(test[2]), key)
+                        self.assertEqual(pt, test[1])
+
+        def testVerify2(self):
+                # Verify that decryption fails if ciphertext is not as long as
+                # RSA modulus
+                self.assertRaises(ValueError, PKCS.decrypt, '\x00'*127,
+                        self.key1024)
+                self.assertRaises(ValueError, PKCS.decrypt, '\x00'*129,
+                        self.key1024)
+
+                # Verify that decryption fails if there are less then 8 non-zero padding
+                # bytes
+                pt = '\x00\x02' + '\xFF'*7 + '\x00' + '\x45'*118
+                ct = self.key1024.encrypt(pt, 0)[0]
+                ct = '\x00'*(128-len(ct)) + ct
+                self.assertRaises(ValueError, PKCS.decrypt, ct, self.key1024)
+
+        def testEncryptVerify1(self):
+                # Encrypt/Verify messages of length [0..RSAlen-11]
+                # and therefore padding [8..117]
+                for pt_len in xrange(0,128-11+1):
+                    pt = self.rng(pt_len)
+                    ct = PKCS.encrypt(pt, self.key1024, self.rng)
+                    pt2 = PKCS.decrypt(ct, self.key1024)
+                    self.assertEqual(pt,pt2)
+
+
+def get_tests(config={}):
+    tests = []
+    tests += list_test_cases(PKCS1_15_Tests)
+    return tests
+
+if __name__ == '__main__':
+    suite = lambda: unittest.TestSuite(get_tests())
+    unittest.main(defaultTest='suite')
+
+# vim:set ts=4 sw=4 sts=4 expandtab: