##// END OF EJS Templates
fix(imports): fixed circular import problem
fix(imports): fixed circular import problem

File last commit:

r5095:aa627a5f default
r5341:115837d2 tip default
Show More
encrypt2.py
84 lines | 2.7 KiB | text/x-python | PythonLexer
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:
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)