fix(encryptor): use a failsafe mechanism of detecting old algo for encryption to NOT crash the app when switching to fernet

rhodecode/lib/enc_utils.py

@@ -1,49 +1,49 b''
 from rhodecode.lib.str_utils import safe_bytes
 from rhodecode.lib.encrypt import encrypt_data, validate_and_decrypt_data
 from rhodecode.lib.encrypt2 import Encryptor
 
 ALLOWED_ALGOS = ['aes', 'fernet']
 
 
 def get_default_algo():
     import rhodecode
     return rhodecode.CONFIG.get('rhodecode.encrypted_values.algorithm') or 'aes'
 
 
 def encrypt_value(value: bytes, enc_key: bytes, algo: str = ''):
     if not algo:
         # not explicit algo, just use what's set by config
         algo = get_default_algo()
 
     if algo not in ALLOWED_ALGOS:
         ValueError(f'Bad encryption algorithm, should be {ALLOWED_ALGOS}, got: {algo}')
 
     enc_key = safe_bytes(enc_key)
     value = safe_bytes(value)
 
     if algo == 'aes':
         return encrypt_data(value, enc_key=enc_key)
     if algo == 'fernet':
         return Encryptor(enc_key).encrypt(value)
 
     return value
 
 
 def decrypt_value(value: bytes, enc_key: bytes, algo: str = '', strict_mode: bool = False):
+    enc_key = safe_bytes(enc_key)
+    value = safe_bytes(value)
 
     if not algo:
         # not explicit algo, just use what's set by config
-        algo = get_default_algo()
+        algo = Encryptor.detect_enc_algo(value) or get_default_algo()
     if algo not in ALLOWED_ALGOS:
         ValueError(f'Bad encryption algorithm, should be {ALLOWED_ALGOS}, got: {algo}')
 
-    enc_key = safe_bytes(enc_key)
-    value = safe_bytes(value)
     safe = not strict_mode
 
     if algo == 'aes':
         return validate_and_decrypt_data(value, enc_key, safe=safe)
     if algo == 'fernet':
         return Encryptor(enc_key).decrypt(value, safe=safe)
 
     return value

rhodecode/lib/encrypt2.py

@@ -1,84 +1,97 b''
 import os
 import base64
 from cryptography.fernet import Fernet, InvalidToken
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
 
 from rhodecode.lib.str_utils import safe_str
 from rhodecode.lib.exceptions import signature_verification_error
 
 
 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 = f'<{cls.__name__}({content[:16]}...)>'
         return str.__new__(cls, content)
 
 
 class Encryptor(object):
     key_format = b'enc2$salt:{1}$data:{2}'
+
     pref_len = 5  # salt:, data:
 
+    @classmethod
+    def detect_enc_algo(cls, enc_data: bytes):
+        parts = enc_data.split(b'$', 3)
+        if len(parts) != 3:
+            raise ValueError(f'Encrypted Data has invalid format, expected {cls.key_format}, got {parts}')
+
+        if b'enc$aes_hmac$' in enc_data:
+            return 'aes'
+        elif b'enc2$salt' in enc_data:
+            return 'fernet'
+        return None
+
     def __init__(self, enc_key: bytes):
         self.enc_key = enc_key
 
     def b64_encode(self, data):
         return base64.urlsafe_b64encode(data)
 
     def b64_decode(self, data):
         return base64.urlsafe_b64decode(data)
 
     def get_encryptor(self, salt):
         """
         Uses Fernet as encryptor with HMAC signature
         :param salt: random salt used for encrypting the data
         """
         kdf = PBKDF2HMAC(
             algorithm=hashes.SHA512(),
             length=32,
             salt=salt,
             iterations=100000,
             backend=default_backend()
         )
         key = self.b64_encode(kdf.derive(self.enc_key))
         return Fernet(key)
 
     def _get_parts(self, enc_data):
         parts = enc_data.split(b'$', 3)
         if len(parts) != 3:
             raise ValueError(f'Encrypted Data has invalid format, expected {self.key_format}, got {parts}')
         prefix, salt, enc_data = parts
 
         try:
             salt = self.b64_decode(salt[self.pref_len:])
         except TypeError:
             # bad base64
             raise ValueError('Encrypted Data salt invalid format, expected base64 format')
 
         enc_data = enc_data[self.pref_len:]
         return prefix, salt, enc_data
 
     def encrypt(self, data) -> bytes:
         salt = os.urandom(64)
         encryptor = self.get_encryptor(salt)
         enc_data = encryptor.encrypt(data)
         return self.key_format.replace(b'{1}', self.b64_encode(salt)).replace(b'{2}', enc_data)
 
     def decrypt(self, data, safe=True) -> bytes | InvalidDecryptedValue:
         parts = self._get_parts(data)
         salt = parts[1]
         enc_data = parts[2]
         encryptor = self.get_encryptor(salt)
         try:
             return encryptor.decrypt(enc_data)
         except (InvalidToken,):
             decrypt_fail = InvalidDecryptedValue(safe_str(data))
             if safe:
                 return decrypt_fail
             raise signature_verification_error(decrypt_fail)