migrate from PyCrypto to cryptograhy #17

Author: rossengeorgiev

requirements.txt

@@ -5,7 +5,7 @@ gevent==1.1.0
 mock==1.3.0
 nose==1.3.7
 protobuf==2.6.1
-pycrypto==2.6.1
+cryptography==1.3
 PyYAML==3.11
 requests==2.9.1
 vcrpy==1.7.4

setup.py

@@ -12,7 +12,7 @@
     __version__ = f.readline().split('"')[1]
 
 install_requires = [
-    'pycrypto>=2.6.1',
+    'cryptography>=1.3',
     'requests>=2.9.1',
     'vdf>=2.0',
 ]

steam/client/__init__.py

@@ -5,7 +5,7 @@
 gevent.monkey.patch_socket()
 gevent.monkey.patch_ssl()
 
-from Crypto.Hash import SHA
+from steam.core.crypto import sha1_hash
 from eventemitter import EventEmitter
 from steam.enums.emsg import EMsg
 from steam.enums import EResult, EOSType, EPersonaState
@@ -120,7 +120,7 @@ def _handle_update_machine_auth(self, message):
 
             resp.body.filename = message.body.filename
             resp.body.eresult = EResult.OK
-            resp.body.sha_file = SHA.new(message.body.bytes).digest()
+            resp.body.sha_file = sha1_hash(message.body.bytes)
             resp.body.getlasterror = 0
             resp.body.offset = message.body.offset
             resp.body.cubwrote = message.body.cubtowrite
@@ -336,7 +336,7 @@ def auth_code_prompt(is_2fa, code_mismatch):
             message.body.eresult_sentryfile = EResult.FileNotFound
         else:
             message.body.eresult_sentryfile = EResult.OK
-            message.body.sha_sentryfile = SHA.new(sentry).digest()
+            message.body.sha_sentryfile = sha1_hash(sentry)
 
         if auth_code:
             message.body.auth_code = auth_code

steam/client/builtins/web.py

@@ -1,10 +1,9 @@
 """
 Web related features
 """
-from Crypto.Hash import SHA
-from Crypto.Random import new as randombytes
+from binascii import hexlify
 from steam import WebAPI
-from steam.core.crypto import generate_session_key, symmetric_encrypt
+from steam.core.crypto import generate_session_key, symmetric_encrypt, sha1_hash, random_bytes
 from steam.util.web import make_requests_session
 
 
@@ -40,7 +39,7 @@ def get_web_session_cookies(self):
             return None
 
         return {
-            'sessionid': SHA.new(randombytes().read(32)).hexdigest(),
+            'sessionid': hexlify(sha1_hash(random_bytes(32))),
             'steamLogin': resp['authenticateuser']['token'],
             'steamLoginSecure': resp['authenticateuser']['tokensecure'],
         }

steam/core/crypto.py

@@ -1,19 +1,26 @@
 import sys
+from os import urandom as random_bytes
 from struct import pack
 from base64 import b64decode
-from Crypto import Random
-from Crypto.Cipher import PKCS1_OAEP, AES
-from Crypto.PublicKey import RSA
-from Crypto.Hash import HMAC, SHA
-
-public_key = """
+from cryptography.hazmat.primitives.hmac import HMAC
+from cryptography.hazmat.primitives.hashes import Hash, SHA1
+from cryptography.hazmat.primitives.asymmetric.padding import PSS, OAEP, MGF1
+from cryptography.hazmat.primitives.ciphers import Cipher
+from cryptography.hazmat.primitives.ciphers.algorithms import AES
+from cryptography.hazmat.primitives.ciphers.modes import CBC, ECB
+import cryptography.hazmat.backends
+backend = cryptography.hazmat.backends.default_backend()
+
+
+class UniverseKey(object):
+    Public = backend.load_der_public_key(b64decode("""
 MIGdMA0GCSqGSIb3DQEBAQUAA4GLADCBhwKBgQDf7BrWLBBmLBc1OhSwfFkRf53T
 2Ct64+AVzRkeRuh7h3SiGEYxqQMUeYKO6UWiSRKpI2hzic9pobFhRr3Bvr/WARvY
 gdTckPv+T1JzZsuVcNfFjrocejN1oWI0Rrtgt4Bo+hOneoo3S57G9F1fOpn5nsQ6
 6WOiu4gZKODnFMBCiQIBEQ==
-"""
+"""))
 
-BS = AES.block_size
+BS = 16
 pad = lambda s: s + (BS - len(s) % BS) * pack('B', BS - len(s) % BS)
 
 if sys.version_info < (3,):
@@ -29,29 +36,35 @@ def generate_session_key(hmac_secret=b''):
     :return: (session_key, encrypted_session_key) tuple
     :rtype: :class:`tuple`
     """
-    session_key = Random.new().read(32)
-    cipher = PKCS1_OAEP.new(RSA.importKey(b64decode(public_key)))
-    encrypted_session_key = cipher.encrypt(session_key + hmac_secret)
+    session_key = random_bytes(32)
+    encrypted_session_key = UniverseKey.Public.encrypt(session_key + hmac_secret,
+                                                       OAEP(MGF1(SHA1()), SHA1(), None)
+                                                       )
     return (session_key, encrypted_session_key)
 
 def symmetric_encrypt(message, key):
-    iv = Random.new().read(BS)
+    iv = random_bytes(BS)
     return symmetric_encrypt_with_iv(message, key, iv)
 
 def symmetric_encrypt_HMAC(message, key, hmac_secret):
-    random_bytes = Random.new().read(3)
+    prefix = random_bytes(3)
 
-    hmac = HMAC.new(hmac_secret, digestmod=SHA)
-    hmac.update(random_bytes)
+    hmac = HMAC(hmac_secret, SHA1(), backend)
+    hmac.update(prefix)
     hmac.update(message)
 
-    iv = hmac.digest()[:13] + random_bytes
+    iv = hmac.finalize()[:13] + prefix
 
     return symmetric_encrypt_with_iv(message, key, iv)
 
+def symmetric_encrypt_iv(iv, key):
+    encryptor =  Cipher(AES(key), ECB(), backend).encryptor()
+    return encryptor.update(iv) + encryptor.finalize()
+
 def symmetric_encrypt_with_iv(message, key, iv):
-    encrypted_iv = AES.new(key, AES.MODE_ECB).encrypt(iv)
-    cyphertext = AES.new(key, AES.MODE_CBC, iv).encrypt(pad(message))
+    encrypted_iv = symmetric_encrypt_iv(iv, key)
+    encryptor =  Cipher(AES(key), CBC(iv), backend).encryptor()
+    cyphertext = encryptor.update(pad(message)) + encryptor.finalize()
     return encrypted_iv + cyphertext
 
 def symmetric_decrypt(cyphertext, key):
@@ -63,18 +76,24 @@ def symmetric_decrypt_HMAC(cyphertext, key, hmac_secret):
     iv = symmetric_decrypt_iv(cyphertext, key)
     message = symmetric_decrypt_with_iv(cyphertext, key, iv)
 
-    hmac = HMAC.new(hmac_secret, digestmod=SHA)
+    hmac = HMAC(hmac_secret, SHA1(), backend)
     hmac.update(iv[-3:])
     hmac.update(message)
 
-    if iv[:13] != hmac.digest()[:13]:
+    if iv[:13] != hmac.finalize()[:13]:
         raise RuntimeError("Unable to decrypt message. HMAC does not match.")
 
     return message
 
 def symmetric_decrypt_iv(cyphertext, key):
-    return AES.new(key, AES.MODE_ECB).decrypt(cyphertext[:BS])
+    decryptor =  Cipher(AES(key), ECB(), backend).decryptor()
+    return decryptor.update(cyphertext[:BS]) + decryptor.finalize()
 
 def symmetric_decrypt_with_iv(cyphertext, key, iv):
-    message = AES.new(key, AES.MODE_CBC, iv).decrypt(cyphertext[BS:])
-    return unpad(message)
+    decryptor =  Cipher(AES(key), CBC(iv), backend).decryptor()
+    return unpad(decryptor.update(cyphertext[BS:]) + decryptor.finalize())
+
+def sha1_hash(data):
+    sha = Hash(SHA1(), backend)
+    sha.update(data)
+    return sha.finalize()

steam/webauth.py

@@ -41,8 +41,11 @@
 import time
 import sys
 from base64 import b64encode
-from Crypto.PublicKey import RSA
-from Crypto.Cipher import PKCS1_v1_5
+
+from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicNumbers
+from cryptography.hazmat.primitives.asymmetric.padding import PKCS1v15
+from steam.core.crypto import backend
+
 from steam.util.web import make_requests_session
 from steam import SteamID
 
@@ -53,7 +56,7 @@
 
 
 class WebAuth(object):
-    cipher = None
+    key = None
     complete = False  #: whether authentication has been completed successfully
     session = None    #: :class:`requests.Session` (with auth cookies after auth is complete)
     captcha_gid = -1
@@ -93,15 +96,15 @@ def get_rsa_key(self, username):
 
         return resp
 
-    def _make_cipher(self):
-        if not self.cipher:
+    def _load_key(self):
+        if not self.key:
             resp = self.get_rsa_key(self.username)
 
-            rsa = RSA.construct((intBase(resp['publickey_mod'], 16),
-                                 intBase(resp['publickey_exp'], 16),
-                                 ))
+            nums = RSAPublicNumbers(intBase(resp['publickey_exp'], 16),
+                                    intBase(resp['publickey_mod'], 16),
+                                    )
 
-            self.cipher = PKCS1_v1_5.new(rsa)
+            self.key = backend.load_rsa_public_numbers(nums)
             self.timestamp = resp['timestamp']
 
     def login(self, captcha='', email_code='', twofactor_code='', language='english'):
@@ -126,11 +129,11 @@ def login(self, captcha='', email_code='', twofactor_code='', language='english'
         if self.complete:
             return self.session
 
-        self._make_cipher()
+        self._load_key()
 
         params = {
             'username' : self.username,
-            "password": b64encode(self.cipher.encrypt(self.password)),
+            "password": b64encode(self.key.encrypt(self.password, PKCS1v15())),
             "emailauth": email_code,
             "emailsteamid": str(self.steamid) if email_code else '',
             "twofactorcode": twofactor_code,

tests/test_core_crypto.py

@@ -7,18 +7,15 @@
 
 class crypto_testcase(unittest.TestCase):
     def setUp(self):
-        class NotRandom:
-            def read(self, n):
-                return b'1' * n
+        patcher = mock.patch('os.urandom')
+        self.addCleanup(patcher.stop)
+        self.urandom = patcher.start()
+        self.urandom.side_effect = lambda n: b'1' * n
 
-        def fakeNew():
-            return NotRandom()
-
-        self._oldnew = crypto.Random.new
-        crypto.Random.new = fakeNew
-
-    def tearDown(self):
-        crypto.Random.new = self._oldnew
+        patcher = mock.patch('steam.core.crypto.random_bytes')
+        self.addCleanup(patcher.stop)
+        self.random_bytes = patcher.start()
+        self.random_bytes.side_effect = lambda n: b'1' * n
 
     def test_keygen(self):
         expected_key = b'1' * 32
@@ -70,4 +67,6 @@ def test_encryption(self):
         with self.assertRaises(RuntimeError):
             crypto.symmetric_decrypt_HMAC(cyphertext, key, b'4'*16)
 
-
+    def test_sha1_hash(self):
+        self.assertEqual(crypto.sha1_hash(b'123'),    b'@\xbd\x00\x15c\x08_\xc3Qe2\x9e\xa1\xff\\^\xcb\xdb\xbe\xef')
+        self.assertEqual(crypto.sha1_hash(b'999999'), b'\x1fU#\xa8\xf55(\x9b4\x01\xb2\x99X\xd0\x1b)f\xeda\xd2')