Use PyCryptodome instead of PyCrypto

Author: bit

requirements.txt

@@ -1 +1 @@
-PyCrypto >= 2.1
+PyCryptodome

src/potr/compatcrypto/pycrypto.py

@@ -15,17 +15,15 @@
 #    You should have received a copy of the GNU Lesser General Public License
 #    along with this library.  If not, see <http://www.gnu.org/licenses/>.
 
-try:
-  import Crypto
-except ImportError:
-  import crypto as Crypto
-
-from Crypto import Cipher
-from Crypto.Hash import SHA256 as _SHA256
-from Crypto.Hash import SHA as _SHA1
-from Crypto.Hash import HMAC as _HMAC
-from Crypto.PublicKey import DSA
-import Crypto.Random.random
+from Cryptodome import Cipher
+from Cryptodome.Hash import HMAC as _HMAC
+from Cryptodome.Hash import SHA256 as _SHA256
+from Cryptodome.Hash import SHA as _SHA1
+from Cryptodome.PublicKey import DSA
+from Cryptodome.Random import random
+from Cryptodome.Signature import DSS
+from Cryptodome.Util import Counter
+
 from numbers import Number
 
 from potr.compatcrypto import common
@@ -45,36 +43,14 @@ def SHA256HMAC(key, data):
 
 def AESCTR(key, counter=0):
     if isinstance(counter, Number):
-        counter = Counter(counter)
-    if not isinstance(counter, Counter):
+        counter = Counter.new(nbits=64, prefix=long_to_bytes(counter, 8), initial_value=0)
+    # in pycrypto Counter used to be an object,
+    # in pycryptodome it's now only a dict.
+    # This tries to validate its "type" so we don't feed anything as a counter
+    if set(counter) != set(Counter.new(64)):
         raise TypeError
     return Cipher.AES.new(key, Cipher.AES.MODE_CTR, counter=counter)
 
-class Counter(object):
-    def __init__(self, prefix):
-        self.prefix = prefix
-        self.val = 0
-
-    def inc(self):
-        self.prefix += 1
-        self.val = 0
-
-    def __setattr__(self, attr, val):
-        if attr == 'prefix':
-            self.val = 0
-        super(Counter, self).__setattr__(attr, val)
-
-    def __repr__(self):
-        return '<Counter(p={p!r},v={v!r})>'.format(p=self.prefix, v=self.val)
-
-    def byteprefix(self):
-        return long_to_bytes(self.prefix, 8)
-
-    def __call__(self):
-        bytesuffix = long_to_bytes(self.val, 8)
-        self.val += 1
-        return self.byteprefix() + bytesuffix
-
 @common.registerkeytype
 class DSAKey(common.PK):
     keyType = 0x0000
@@ -107,12 +83,14 @@ def fingerprint(self):
     def sign(self, data):
         # 2 <= K <= q
         K = randrange(2, self.priv.q)
-        r, s = self.priv.sign(data, K)
+        M = bytes_to_long(data)
+        r, s = self.priv._sign(M, K)
         return long_to_bytes(r, 20) + long_to_bytes(s, 20)
 
     def verify(self, data, sig):
         r, s = bytes_to_long(sig[:20]), bytes_to_long(sig[20:])
-        return self.pub.verify(data, (r, s))
+        M = bytes_to_long(data)
+        return self.pub._verify(M, (r, s))
 
     def __hash__(self):
         return bytes_to_long(self.fingerprint())
@@ -128,8 +106,8 @@ def __ne__(self, other):
     @classmethod
     def generate(cls):
         privkey = DSA.generate(1024)
-        return cls((privkey.key.y, privkey.key.g, privkey.key.p, privkey.key.q,
-                privkey.key.x), private=True)
+        return cls((privkey.y, privkey.g, privkey.p, privkey.q,
+                privkey.x), private=True)
 
     @classmethod
     def parsePayload(cls, data, private=False):
@@ -143,7 +121,7 @@ def parsePayload(cls, data, private=False):
         return cls((y, g, p, q), private=False), data
 
 def getrandbits(k):
-    return Crypto.Random.random.getrandbits(k)
+    return random.getrandbits(k)
 
 def randrange(start, stop):
-    return Crypto.Random.random.randrange(start, stop)
+    return random.randrange(start, stop)

src/potr/crypt.py

@@ -23,7 +23,7 @@
 
 
 from potr.compatcrypto import SHA256, SHA1, SHA1HMAC, SHA256HMAC, \
-        Counter, AESCTR, PK, getrandbits, randrange
+        AESCTR, PK, getrandbits, randrange
 from potr.utils import bytes_to_long, long_to_bytes, pack_mpi, read_mpi
 from potr import proto
 
@@ -69,8 +69,8 @@ def __init__(self, sendenc, sendmac, rcvenc, rcvmac):
         self.sendmac = sendmac
         self.rcvenc = rcvenc
         self.rcvmac = rcvmac
-        self.sendctr = Counter(0)
-        self.rcvctr = Counter(0)
+        self.sendctr = 0
+        self.rcvctr = 0
         self.sendmacused = False
         self.rcvmacused = False
 
@@ -177,12 +177,12 @@ def handleDataMessage(self, msg):
         sesskey.rcvmacused = True
 
         newCtrPrefix = bytes_to_long(msg.ctr)
-        if newCtrPrefix <= sesskey.rcvctr.prefix:
+        if newCtrPrefix <= sesskey.rcvctr:
             logger.error('CTR must increase (old %r, new %r)',
                     sesskey.rcvctr.prefix, newCtrPrefix)
             raise InvalidParameterError
 
-        sesskey.rcvctr.prefix = newCtrPrefix
+        sesskey.rcvctr = newCtrPrefix
 
         logger.debug('handle: enc={0!r} mac={1!r} ctr={2!r}' \
                 .format(sesskey.rcvenc, sesskey.rcvmac, sesskey.rcvctr))
@@ -232,7 +232,7 @@ def createDataMessage(self, message, flags=0, tlvs=None):
             tlvs = []
 
         sess = self.sessionkeys[1][0]
-        sess.sendctr.inc()
+        sess.sendctr += 1
 
         logger.debug('create: enc={0!r} mac={1!r} ctr={2!r}' \
                 .format(sess.sendenc, sess.sendmac, sess.sendctr))
@@ -242,7 +242,7 @@ def createDataMessage(self, message, flags=0, tlvs=None):
         encmsg = AESCTR(sess.sendenc, sess.sendctr).encrypt(plainBuf)
 
         msg = proto.DataMessage(flags, self.ourKeyid-1, self.theirKeyid,
-                long_to_bytes(self.ourDHKey.pub), sess.sendctr.byteprefix(),
+                long_to_bytes(self.ourDHKey.pub), long_to_bytes(sess.sendctr, 8),
                 encmsg, b'', b''.join(self.savedMacKeys))
 
         self.savedMacKeys = []

tests/test_compatcrypto.py

@@ -64,10 +64,10 @@ def test_AESCTR_counter_counter(self):
         key = potr.utils.long_to_bytes(
             potr.compatcrypto.getrandbits(128), 16)
 
-        aes_encrypter = potr.compatcrypto.AESCTR(key, potr.compatcrypto.Counter(2013))
+        aes_encrypter = potr.compatcrypto.AESCTR(key, 2013)
         ciphertext = aes_encrypter.encrypt(b'setec astronomy')
 
-        aes_decrypter = potr.compatcrypto.AESCTR(key, potr.compatcrypto.Counter(2013))
+        aes_decrypter = potr.compatcrypto.AESCTR(key, 2013)
         self.assertEqual(aes_decrypter.decrypt(ciphertext), b'setec astronomy')
 
     def test_getrandbits(self):