encrypt-module: fixed str passed to function that expected bytes

rhodecode/lib/encrypt.py

@@ -1,138 +1,138 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2014-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 """
 Generic encryption library for RhodeCode
 """
 
 import base64
 
 from Crypto.Cipher import AES
 from Crypto import Random
 from Crypto.Hash import HMAC, SHA256
 
-from rhodecode.lib.utils2 import safe_str
+from rhodecode.lib.str_utils import safe_bytes
 
 
 class SignatureVerificationError(Exception):
     pass
 
 
 class InvalidDecryptedValue(str):
 
     def __new__(cls, content):
         """
         This will generate something like this::
             <InvalidDecryptedValue(QkWusFgLJXR6m42v...)>
         And represent a safe indicator that encryption key is broken
         """
         content = '<{}({}...)>'.format(cls.__name__, content[:16])
         return str.__new__(cls, content)
 
 
 class AESCipher(object):
     def __init__(self, key, hmac=False, strict_verification=True):
         if not key:
             raise ValueError('passed key variable is empty')
         self.strict_verification = strict_verification
         self.block_size = 32
         self.hmac_size = 32
         self.hmac = hmac
 
-        self.key = SHA256.new(safe_str(key)).digest()
+        self.key = SHA256.new(safe_bytes(key)).digest()
         self.hmac_key = SHA256.new(self.key).digest()
 
     def verify_hmac_signature(self, raw_data):
         org_hmac_signature = raw_data[-self.hmac_size:]
         data_without_sig = raw_data[:-self.hmac_size]
         recomputed_hmac = HMAC.new(
             self.hmac_key, data_without_sig, digestmod=SHA256).digest()
         return org_hmac_signature == recomputed_hmac
 
     def encrypt(self, raw):
         raw = self._pad(raw)
         iv = Random.new().read(AES.block_size)
         cipher = AES.new(self.key, AES.MODE_CBC, iv)
         enc_value = cipher.encrypt(raw)
 
         hmac_signature = ''
         if self.hmac:
             # compute hmac+sha256 on iv + enc text, we use
             # encrypt then mac method to create the signature
             hmac_signature = HMAC.new(
                 self.hmac_key, iv + enc_value, digestmod=SHA256).digest()
 
         return base64.b64encode(iv + enc_value + hmac_signature)
 
     def decrypt(self, enc):
         enc_org = enc
         enc = base64.b64decode(enc)
 
         if self.hmac and len(enc) > self.hmac_size:
             if self.verify_hmac_signature(enc):
                 # cut off the HMAC verification digest
                 enc = enc[:-self.hmac_size]
             else:
                 if self.strict_verification:
                     raise SignatureVerificationError(
                         "Encryption signature verification failed. "
                         "Please check your secret key, and/or encrypted value. "
                         "Secret key is stored as "
                         "`rhodecode.encrypted_values.secret` or "
                         "`beaker.session.secret` inside .ini file")
 
                 return InvalidDecryptedValue(enc_org)
 
         iv = enc[:AES.block_size]
         cipher = AES.new(self.key, AES.MODE_CBC, iv)
         return self._unpad(cipher.decrypt(enc[AES.block_size:]))
 
     def _pad(self, s):
         return (s + (self.block_size - len(s) % self.block_size)
                 * chr(self.block_size - len(s) % self.block_size))
 
     @staticmethod
     def _unpad(s):
         return s[:-ord(s[len(s)-1:])]
 
 
 def validate_and_get_enc_data(enc_data, enc_key, enc_strict_mode):
     parts = enc_data.split('$', 3)
     if not len(parts) == 3:
         # probably not encrypted values
         return enc_data
     else:
         if parts[0] != 'enc':
             # parts ok but without our header ?
             return enc_data
 
         # at that stage we know it's our encryption
         if parts[1] == 'aes':
             decrypted_data = AESCipher(enc_key).decrypt(parts[2])
         elif parts[1] == 'aes_hmac':
             decrypted_data = AESCipher(
                 enc_key, hmac=True,
                 strict_verification=enc_strict_mode).decrypt(parts[2])
         else:
             raise ValueError(
                 'Encryption type part is wrong, must be `aes` '
                 'or `aes_hmac`, got `%s` instead' % (parts[1]))
         return decrypted_data