edq.util.crypto

View Source

  1import enum
  2import hashlib
  3import logging
  4import os
  5import typing
  6
  7import cryptography.hazmat.primitives
  8import cryptography.hazmat.primitives.ciphers
  9import cryptography.hazmat.primitives.padding
 10
 11import edq.core.errors
 12import edq.util.constants
 13import edq.util.encoding
 14import edq.util.enum
 15import edq.util.serial
 16
 17_logger = logging.getLogger(__name__)
 18
 19SCRYPT_DEFAULT_ITERATIONS: int = 2**11
 20"""
 21The default iterations to use with scrypt.
 222^11 or 2^14 are generally recommended for interactive tools.
 23"""
 24
 25SCRYPT_DEFAULT_BLOCK_SIZE: int = 8
 26""" The default block size factor to use with scrypt. """
 27
 28SCRYPT_PARALLELIZATION: int = 1
 29""" The parallelization (or lack thereof) to use in scrypt. """
 30
 31SALT_LENGTH_BYTES: int = 16
 32""" The length for generated salts. """
 33
 34IV_LENGTH_BYTES: int = 16
 35""" The length for generated IVs (before key derivation). """
 36
 37AES_BLOCK_SIZE_BYTES: int = 16
 38""" AES always uses 128-bit blocks. """
 39
 40SECRET_PREFIX: str = '__edq_secret__'
 41""" A string that must appear at the beginning of an encrypted secret. """
 42
 43SECRET_DELIM: str = '::'
 44""" The delimiter for the components of an encrypted secret. """
 45
 46class EncryptionMethod(enum.Enum):
 47    """ Supported encryption methods. """
 48
 49    AES256v1 = 'AES256v1'  # pylint: disable=invalid-name
 50
 51class Secret(edq.util.serial.PODConverter):
 52    """
 53    Secrets represent data that should be protected on disk.
 54    This type can be configured to always be serialized in an encrypted form.
 55    """
 56
 57    def __init__(self,
 58            cleartext: str,
 59            iv_b64: typing.Union[str, None] = None,
 60            salt_b64: typing.Union[str, None] = None,
 61            encryption_method: EncryptionMethod = EncryptionMethod.AES256v1,
 62            write_encrypted: typing.Union[bool, None] = None,
 63            ) -> None:
 64        self.cleartext: str = cleartext
 65        """ The contexts of the secret. """
 66
 67        self.iv_b64: typing.Union[str, None] = iv_b64
 68        """
 69        The base64 of the iv to use during encryption.
 70        May be none is this secret has not been encrypted yet,
 71        will be set during the first encryption.
 72        """
 73
 74        self.salt_b64: typing.Union[str, None] = salt_b64
 75        """
 76        The base64 of the salt to use during encryption.
 77        May be none is this secret has not been encrypted yet,
 78        will be set during the first encryption.
 79        """
 80
 81        self.encryption_method: EncryptionMethod = encryption_method
 82        """ The encryption method to use. """
 83
 84        if (write_encrypted is None):
 85            write_encrypted = (salt_b64 is not None)
 86
 87        self.write_encrypted: bool = write_encrypted
 88        """
 89        Whether to write (serialize) this secret in its encrypted form.
 90        If not explicitly set, this will be set based on if the passed in salt is null
 91        (if the salt not null, then encrypt).
 92        """
 93
 94    def is_encrypted(self) -> bool:
 95        """
 96        Check if this secret is encrypted on disk.
 97        This is meant to be used by regular users who use parse().
 98        Advanced users may have to make their own checks.
 99        """
100
101        return (self.salt_b64 is not None)
102
103    def __repr__(self) -> str:
104        return self.cleartext
105
106    # Use generic dict serialization for equality.
107    def __eq__(self, other: object) -> bool:
108        if (not isinstance(other, Secret)):
109            return False
110
111        context = edq.util.serial.SerializationContext()
112        return bool(super().to_pod(context) == super(edq.util.serial.PODConverter, other).to_pod(context))  # type: ignore[attr-defined,unused-ignore]
113
114    def encrypt(self, key: str) -> str:
115        """
116        Get the full delimited and encrypted representation for this secret.
117        If the IV and/or salt has not been set, this call will set them.
118        """
119
120        if (self.encryption_method == EncryptionMethod.AES256v1):
121            ciphertext, self.iv_b64, self.salt_b64 = aes256_encrypt(key, self.cleartext, self.iv_b64, self.salt_b64)
122        else:
123            raise edq.core.errors.SerializationError(f"Secret has an unsupported encryption method: '{self.encryption_method}'.")
124
125        parts: typing.List[str] = [
126            SECRET_PREFIX,
127            self.encryption_method.value,
128            self.iv_b64,
129            self.salt_b64,
130            ciphertext,
131        ]
132
133        return SECRET_DELIM.join(parts)
134
135    def to_pod(self,
136            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
137            ) -> edq.util.serial.PODType:
138        if (context is None):
139            context = edq.util.serial.SerializationContext()
140
141        if (not self.write_encrypted):
142            return self.cleartext
143
144        if (context.key is None):
145            raise edq.core.errors.SerializationError("No key provided for writing an encrypted secret.")
146
147        return self.encrypt(context.key)
148
149    @classmethod
150    def parse(cls, text: str, key: typing.Union[str, None] = None) -> 'Secret':
151        """
152        Parse a secret from text.
153        A key is required if the text is encrypted.
154        """
155
156        text = text.strip()
157
158        # Check for only cleartext.
159        if (not text.startswith(SECRET_PREFIX)):
160            return Secret(text)
161
162        if (key is None):
163            raise edq.core.errors.SerializationError("No key provided for reading an encrypted secret.")
164
165        parts = text.split(SECRET_DELIM)
166        if (len(parts) != 5):
167            raise edq.core.errors.SerializationError(f"Secret has an unexpected number of parts, expecting 5 and found {len(parts)}: '{text}'.")
168
169        (_, raw_method, iv_b64, salt_b64, ciphertext_b64) = parts
170
171        if (not edq.util.enum.has_value(EncryptionMethod, raw_method)):
172            raise edq.core.errors.SerializationError(f"Secret has an unknown encryption method: '{raw_method}'.")
173
174        encryption_method = EncryptionMethod(raw_method)
175
176        if (encryption_method == EncryptionMethod.AES256v1):
177            cleartext = aes256_decrypt(key, iv_b64, salt_b64, ciphertext_b64)
178        else:
179            raise edq.core.errors.SerializationError(f"Secret has an unsupported encryption method: '{encryption_method}'.")
180
181        return Secret(cleartext, iv_b64, salt_b64, encryption_method)
182
183    @classmethod
184    def from_pod(cls,
185            data: edq.util.serial.PODType,
186            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
187            ) -> 'Secret':
188        if (not isinstance(data, str)):
189            raise edq.core.errors.SerializationError(f"Secrets should be deserialized from a string, found a '{type(data)}' ({data}).")
190
191        if (context is None):
192            context = edq.util.serial.SerializationContext()
193
194        return cls.parse(str(data), context.key)
195
196def aes256_encrypt(
197        key: str,
198        cleartext: str,
199        iv_b64: typing.Union[str, None] = None,
200        salt_b64: typing.Union[str, None] = None,
201        encoding: str = edq.util.constants.DEFAULT_ENCODING,
202        ) -> typing.Tuple[str, str, str]:
203    """
204    Perform AES256-CBC encryption.
205    Returns the base64 encoding of the ciphertext, iv, and salt.
206    If the IV and/or salt are passed in, the same values will be passed back.
207    """
208
209    # Resolve optional fields.
210
211    if (iv_b64 is None):
212        iv_bytes = os.urandom(IV_LENGTH_BYTES)
213    else:
214        iv_bytes = edq.util.encoding.from_base64(iv_b64, encoding = encoding)
215
216    if (salt_b64 is None):
217        salt_bytes = os.urandom(SALT_LENGTH_BYTES)
218    else:
219        salt_bytes = edq.util.encoding.from_base64(salt_b64, encoding = encoding)
220
221    # Derive fixed-sized keys from the key (32 bytes) and IV (16 bytes).
222    aes_key = _derive_aes_key(key.encode(encoding), salt_bytes, 32)
223    aes_iv = _derive_aes_key(iv_bytes, salt_bytes, AES_BLOCK_SIZE_BYTES)
224
225    # Convert the cleartext to bytes and pad for a 256 block size.
226    cleartext_bytes = cleartext.encode(encoding)
227    padder = cryptography.hazmat.primitives.padding.PKCS7(8 * AES_BLOCK_SIZE_BYTES).padder()
228    cleartext_bytes_padded = padder.update(cleartext_bytes) + padder.finalize()
229
230    # Encrypt the data.
231    cipher = cryptography.hazmat.primitives.ciphers.Cipher(
232        cryptography.hazmat.primitives.ciphers.algorithms.AES(aes_key),
233        cryptography.hazmat.primitives.ciphers.modes.CBC(aes_iv),
234    )
235    encryptor = cipher.encryptor()
236    ciphertext_bytes = encryptor.update(cleartext_bytes_padded) + encryptor.finalize()
237
238    # Encode the results in base64.
239    ciphertext_b64 = edq.util.encoding.to_base64(ciphertext_bytes)
240    iv_b64 = edq.util.encoding.to_base64(iv_bytes)
241    salt_b64 = edq.util.encoding.to_base64(salt_bytes)
242
243    return (ciphertext_b64, iv_b64, salt_b64)
244
245def aes256_decrypt(
246        key: str,
247        iv_b64: str,
248        salt_b64: str,
249        ciphertext_b64: str,
250        encoding: str = edq.util.constants.DEFAULT_ENCODING,
251        ) -> str:
252    """
253    Perform AES256-CBC decryption.
254    """
255
256    # Get the input data as bytes.
257    iv_bytes = edq.util.encoding.from_base64(iv_b64, encoding = encoding)
258    salt_bytes = edq.util.encoding.from_base64(salt_b64, encoding = encoding)
259    ciphertext_bytes = edq.util.encoding.from_base64(ciphertext_b64, encoding = encoding)
260
261    # Derive fixed-sized keys from the key (32 bytes) and IV (16 bytes).
262    aes_key = _derive_aes_key(key.encode(encoding), salt_bytes, 32)
263    aes_iv = _derive_aes_key(iv_bytes, salt_bytes, AES_BLOCK_SIZE_BYTES)
264
265    # Decrypt the data.
266    cipher = cryptography.hazmat.primitives.ciphers.Cipher(
267        cryptography.hazmat.primitives.ciphers.algorithms.AES(aes_key),
268        cryptography.hazmat.primitives.ciphers.modes.CBC(aes_iv),
269    )
270    decryptor = cipher.decryptor()
271    cleartext_bytes_padded = decryptor.update(ciphertext_bytes) + decryptor.finalize()
272
273    # Unpad and decode the data.
274    unpadder = cryptography.hazmat.primitives.padding.PKCS7(8 * AES_BLOCK_SIZE_BYTES).unpadder()
275    try:
276        cleartext_bytes = unpadder.update(cleartext_bytes_padded) + unpadder.finalize()
277    except ValueError as ex:
278        _logger.debug("Decryption failed.", exc_info = ex)
279        raise edq.core.errors.UtilsError("Decryption failed. Do you have the correct key?")  # pylint: disable=raise-missing-from
280
281    cleartext = cleartext_bytes.decode(encoding)
282
283    return cleartext
284
285def _derive_aes_key(
286        key_bytes: bytes,
287        salt: bytes,
288        derived_key_length: int,
289        iterations: int = SCRYPT_DEFAULT_ITERATIONS,
290        block_size: int = SCRYPT_DEFAULT_BLOCK_SIZE,
291        ) -> bytes:
292    """ Derive a key for use in AES256 from a cleartext key and salt. """
293
294    return hashlib.scrypt(
295        key_bytes,
296        salt = salt,
297        n = iterations,
298        r = block_size,
299        p = SCRYPT_PARALLELIZATION,
300        dklen = derived_key_length,
301    )

SCRYPT_DEFAULT_ITERATIONS: int = 2048

The default iterations to use with scrypt. 2^11 or 2^14 are generally recommended for interactive tools.

SCRYPT_DEFAULT_BLOCK_SIZE: int = 8

The default block size factor to use with scrypt.

SCRYPT_PARALLELIZATION: int = 1

The parallelization (or lack thereof) to use in scrypt.

SALT_LENGTH_BYTES: int = 16

The length for generated salts.

IV_LENGTH_BYTES: int = 16

The length for generated IVs (before key derivation).

AES_BLOCK_SIZE_BYTES: int = 16

AES always uses 128-bit blocks.

SECRET_PREFIX: str = '__edq_secret__'

A string that must appear at the beginning of an encrypted secret.

SECRET_DELIM: str = '::'

The delimiter for the components of an encrypted secret.

class EncryptionMethod(enum.Enum): View Source

47class EncryptionMethod(enum.Enum):
48    """ Supported encryption methods. """
49
50    AES256v1 = 'AES256v1'  # pylint: disable=invalid-name

Supported encryption methods.

AES256v1 = <EncryptionMethod.AES256v1: 'AES256v1'>

class Secret(edq.util.serial.PODConverter): View Source

 52class Secret(edq.util.serial.PODConverter):
 53    """
 54    Secrets represent data that should be protected on disk.
 55    This type can be configured to always be serialized in an encrypted form.
 56    """
 57
 58    def __init__(self,
 59            cleartext: str,
 60            iv_b64: typing.Union[str, None] = None,
 61            salt_b64: typing.Union[str, None] = None,
 62            encryption_method: EncryptionMethod = EncryptionMethod.AES256v1,
 63            write_encrypted: typing.Union[bool, None] = None,
 64            ) -> None:
 65        self.cleartext: str = cleartext
 66        """ The contexts of the secret. """
 67
 68        self.iv_b64: typing.Union[str, None] = iv_b64
 69        """
 70        The base64 of the iv to use during encryption.
 71        May be none is this secret has not been encrypted yet,
 72        will be set during the first encryption.
 73        """
 74
 75        self.salt_b64: typing.Union[str, None] = salt_b64
 76        """
 77        The base64 of the salt to use during encryption.
 78        May be none is this secret has not been encrypted yet,
 79        will be set during the first encryption.
 80        """
 81
 82        self.encryption_method: EncryptionMethod = encryption_method
 83        """ The encryption method to use. """
 84
 85        if (write_encrypted is None):
 86            write_encrypted = (salt_b64 is not None)
 87
 88        self.write_encrypted: bool = write_encrypted
 89        """
 90        Whether to write (serialize) this secret in its encrypted form.
 91        If not explicitly set, this will be set based on if the passed in salt is null
 92        (if the salt not null, then encrypt).
 93        """
 94
 95    def is_encrypted(self) -> bool:
 96        """
 97        Check if this secret is encrypted on disk.
 98        This is meant to be used by regular users who use parse().
 99        Advanced users may have to make their own checks.
100        """
101
102        return (self.salt_b64 is not None)
103
104    def __repr__(self) -> str:
105        return self.cleartext
106
107    # Use generic dict serialization for equality.
108    def __eq__(self, other: object) -> bool:
109        if (not isinstance(other, Secret)):
110            return False
111
112        context = edq.util.serial.SerializationContext()
113        return bool(super().to_pod(context) == super(edq.util.serial.PODConverter, other).to_pod(context))  # type: ignore[attr-defined,unused-ignore]
114
115    def encrypt(self, key: str) -> str:
116        """
117        Get the full delimited and encrypted representation for this secret.
118        If the IV and/or salt has not been set, this call will set them.
119        """
120
121        if (self.encryption_method == EncryptionMethod.AES256v1):
122            ciphertext, self.iv_b64, self.salt_b64 = aes256_encrypt(key, self.cleartext, self.iv_b64, self.salt_b64)
123        else:
124            raise edq.core.errors.SerializationError(f"Secret has an unsupported encryption method: '{self.encryption_method}'.")
125
126        parts: typing.List[str] = [
127            SECRET_PREFIX,
128            self.encryption_method.value,
129            self.iv_b64,
130            self.salt_b64,
131            ciphertext,
132        ]
133
134        return SECRET_DELIM.join(parts)
135
136    def to_pod(self,
137            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
138            ) -> edq.util.serial.PODType:
139        if (context is None):
140            context = edq.util.serial.SerializationContext()
141
142        if (not self.write_encrypted):
143            return self.cleartext
144
145        if (context.key is None):
146            raise edq.core.errors.SerializationError("No key provided for writing an encrypted secret.")
147
148        return self.encrypt(context.key)
149
150    @classmethod
151    def parse(cls, text: str, key: typing.Union[str, None] = None) -> 'Secret':
152        """
153        Parse a secret from text.
154        A key is required if the text is encrypted.
155        """
156
157        text = text.strip()
158
159        # Check for only cleartext.
160        if (not text.startswith(SECRET_PREFIX)):
161            return Secret(text)
162
163        if (key is None):
164            raise edq.core.errors.SerializationError("No key provided for reading an encrypted secret.")
165
166        parts = text.split(SECRET_DELIM)
167        if (len(parts) != 5):
168            raise edq.core.errors.SerializationError(f"Secret has an unexpected number of parts, expecting 5 and found {len(parts)}: '{text}'.")
169
170        (_, raw_method, iv_b64, salt_b64, ciphertext_b64) = parts
171
172        if (not edq.util.enum.has_value(EncryptionMethod, raw_method)):
173            raise edq.core.errors.SerializationError(f"Secret has an unknown encryption method: '{raw_method}'.")
174
175        encryption_method = EncryptionMethod(raw_method)
176
177        if (encryption_method == EncryptionMethod.AES256v1):
178            cleartext = aes256_decrypt(key, iv_b64, salt_b64, ciphertext_b64)
179        else:
180            raise edq.core.errors.SerializationError(f"Secret has an unsupported encryption method: '{encryption_method}'.")
181
182        return Secret(cleartext, iv_b64, salt_b64, encryption_method)
183
184    @classmethod
185    def from_pod(cls,
186            data: edq.util.serial.PODType,
187            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
188            ) -> 'Secret':
189        if (not isinstance(data, str)):
190            raise edq.core.errors.SerializationError(f"Secrets should be deserialized from a string, found a '{type(data)}' ({data}).")
191
192        if (context is None):
193            context = edq.util.serial.SerializationContext()
194
195        return cls.parse(str(data), context.key)

Secrets represent data that should be protected on disk. This type can be configured to always be serialized in an encrypted form.

Secret( cleartext: str, iv_b64: Optional[str] = None, salt_b64: Optional[str] = None, encryption_method: EncryptionMethod = <EncryptionMethod.AES256v1: 'AES256v1'>, write_encrypted: Optional[bool] = None) View Source

58    def __init__(self,
59            cleartext: str,
60            iv_b64: typing.Union[str, None] = None,
61            salt_b64: typing.Union[str, None] = None,
62            encryption_method: EncryptionMethod = EncryptionMethod.AES256v1,
63            write_encrypted: typing.Union[bool, None] = None,
64            ) -> None:
65        self.cleartext: str = cleartext
66        """ The contexts of the secret. """
67
68        self.iv_b64: typing.Union[str, None] = iv_b64
69        """
70        The base64 of the iv to use during encryption.
71        May be none is this secret has not been encrypted yet,
72        will be set during the first encryption.
73        """
74
75        self.salt_b64: typing.Union[str, None] = salt_b64
76        """
77        The base64 of the salt to use during encryption.
78        May be none is this secret has not been encrypted yet,
79        will be set during the first encryption.
80        """
81
82        self.encryption_method: EncryptionMethod = encryption_method
83        """ The encryption method to use. """
84
85        if (write_encrypted is None):
86            write_encrypted = (salt_b64 is not None)
87
88        self.write_encrypted: bool = write_encrypted
89        """
90        Whether to write (serialize) this secret in its encrypted form.
91        If not explicitly set, this will be set based on if the passed in salt is null
92        (if the salt not null, then encrypt).
93        """

cleartext: str

The contexts of the secret.

iv_b64: Optional[str]

The base64 of the iv to use during encryption. May be none is this secret has not been encrypted yet, will be set during the first encryption.

salt_b64: Optional[str]

The base64 of the salt to use during encryption. May be none is this secret has not been encrypted yet, will be set during the first encryption.

encryption_method: EncryptionMethod

The encryption method to use.

write_encrypted: bool

Whether to write (serialize) this secret in its encrypted form. If not explicitly set, this will be set based on if the passed in salt is null (if the salt not null, then encrypt).

def is_encrypted(self) -> bool: View Source

 95    def is_encrypted(self) -> bool:
 96        """
 97        Check if this secret is encrypted on disk.
 98        This is meant to be used by regular users who use parse().
 99        Advanced users may have to make their own checks.
100        """
101
102        return (self.salt_b64 is not None)

Check if this secret is encrypted on disk. This is meant to be used by regular users who use parse(). Advanced users may have to make their own checks.

def encrypt(self, key: str) -> str: View Source

115    def encrypt(self, key: str) -> str:
116        """
117        Get the full delimited and encrypted representation for this secret.
118        If the IV and/or salt has not been set, this call will set them.
119        """
120
121        if (self.encryption_method == EncryptionMethod.AES256v1):
122            ciphertext, self.iv_b64, self.salt_b64 = aes256_encrypt(key, self.cleartext, self.iv_b64, self.salt_b64)
123        else:
124            raise edq.core.errors.SerializationError(f"Secret has an unsupported encryption method: '{self.encryption_method}'.")
125
126        parts: typing.List[str] = [
127            SECRET_PREFIX,
128            self.encryption_method.value,
129            self.iv_b64,
130            self.salt_b64,
131            ciphertext,
132        ]
133
134        return SECRET_DELIM.join(parts)

Get the full delimited and encrypted representation for this secret. If the IV and/or salt has not been set, this call will set them.

def to_pod( self, context: Optional[edq.util.common.SerializationContext] = None) -> Union[bool, float, int, str, List[ForwardRef('PODType')], Dict[str, ForwardRef('PODType')], NoneType]: View Source

136    def to_pod(self,
137            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
138            ) -> edq.util.serial.PODType:
139        if (context is None):
140            context = edq.util.serial.SerializationContext()
141
142        if (not self.write_encrypted):
143            return self.cleartext
144
145        if (context.key is None):
146            raise edq.core.errors.SerializationError("No key provided for writing an encrypted secret.")
147
148        return self.encrypt(context.key)

Get a POD representation of this object.

The default implementation will convert to a dict (similar to a DictSerializer).

@classmethod

def parse(cls, text: str, key: Optional[str] = None) -> Secret: View Source

150    @classmethod
151    def parse(cls, text: str, key: typing.Union[str, None] = None) -> 'Secret':
152        """
153        Parse a secret from text.
154        A key is required if the text is encrypted.
155        """
156
157        text = text.strip()
158
159        # Check for only cleartext.
160        if (not text.startswith(SECRET_PREFIX)):
161            return Secret(text)
162
163        if (key is None):
164            raise edq.core.errors.SerializationError("No key provided for reading an encrypted secret.")
165
166        parts = text.split(SECRET_DELIM)
167        if (len(parts) != 5):
168            raise edq.core.errors.SerializationError(f"Secret has an unexpected number of parts, expecting 5 and found {len(parts)}: '{text}'.")
169
170        (_, raw_method, iv_b64, salt_b64, ciphertext_b64) = parts
171
172        if (not edq.util.enum.has_value(EncryptionMethod, raw_method)):
173            raise edq.core.errors.SerializationError(f"Secret has an unknown encryption method: '{raw_method}'.")
174
175        encryption_method = EncryptionMethod(raw_method)
176
177        if (encryption_method == EncryptionMethod.AES256v1):
178            cleartext = aes256_decrypt(key, iv_b64, salt_b64, ciphertext_b64)
179        else:
180            raise edq.core.errors.SerializationError(f"Secret has an unsupported encryption method: '{encryption_method}'.")
181
182        return Secret(cleartext, iv_b64, salt_b64, encryption_method)

Parse a secret from text. A key is required if the text is encrypted.

@classmethod

def from_pod( cls, data: Union[bool, float, int, str, List[ForwardRef('PODType')], Dict[str, ForwardRef('PODType')], NoneType], context: Optional[edq.util.common.SerializationContext] = None) -> Secret: View Source

184    @classmethod
185    def from_pod(cls,
186            data: edq.util.serial.PODType,
187            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
188            ) -> 'Secret':
189        if (not isinstance(data, str)):
190            raise edq.core.errors.SerializationError(f"Secrets should be deserialized from a string, found a '{type(data)}' ({data}).")
191
192        if (context is None):
193            context = edq.util.serial.SerializationContext()
194
195        return cls.parse(str(data), context.key)

Create an instance of this class from a POD.

The default implementation will call the class' constructor with one of two things: a splat/unpacking (**) of the incoming data if the data is a dict, otherwise the data itself.

Inherited Members

edq.util.serial.PODConverter: copy
edq.util.serial.PODSerializer: to_path
edq.util.serial.PODDeserializer: prep_init_data; from_path
edq.util.serial.SerializationBase: serialization_omit_empty; serialization_omit_none; serialization_skip_fields; serialization_include_init_context; serialization_error_class; skip_field

def aes256_encrypt( key: str, cleartext: str, iv_b64: Optional[str] = None, salt_b64: Optional[str] = None, encoding: str = 'utf-8') -> Tuple[str, str, str]: View Source

197def aes256_encrypt(
198        key: str,
199        cleartext: str,
200        iv_b64: typing.Union[str, None] = None,
201        salt_b64: typing.Union[str, None] = None,
202        encoding: str = edq.util.constants.DEFAULT_ENCODING,
203        ) -> typing.Tuple[str, str, str]:
204    """
205    Perform AES256-CBC encryption.
206    Returns the base64 encoding of the ciphertext, iv, and salt.
207    If the IV and/or salt are passed in, the same values will be passed back.
208    """
209
210    # Resolve optional fields.
211
212    if (iv_b64 is None):
213        iv_bytes = os.urandom(IV_LENGTH_BYTES)
214    else:
215        iv_bytes = edq.util.encoding.from_base64(iv_b64, encoding = encoding)
216
217    if (salt_b64 is None):
218        salt_bytes = os.urandom(SALT_LENGTH_BYTES)
219    else:
220        salt_bytes = edq.util.encoding.from_base64(salt_b64, encoding = encoding)
221
222    # Derive fixed-sized keys from the key (32 bytes) and IV (16 bytes).
223    aes_key = _derive_aes_key(key.encode(encoding), salt_bytes, 32)
224    aes_iv = _derive_aes_key(iv_bytes, salt_bytes, AES_BLOCK_SIZE_BYTES)
225
226    # Convert the cleartext to bytes and pad for a 256 block size.
227    cleartext_bytes = cleartext.encode(encoding)
228    padder = cryptography.hazmat.primitives.padding.PKCS7(8 * AES_BLOCK_SIZE_BYTES).padder()
229    cleartext_bytes_padded = padder.update(cleartext_bytes) + padder.finalize()
230
231    # Encrypt the data.
232    cipher = cryptography.hazmat.primitives.ciphers.Cipher(
233        cryptography.hazmat.primitives.ciphers.algorithms.AES(aes_key),
234        cryptography.hazmat.primitives.ciphers.modes.CBC(aes_iv),
235    )
236    encryptor = cipher.encryptor()
237    ciphertext_bytes = encryptor.update(cleartext_bytes_padded) + encryptor.finalize()
238
239    # Encode the results in base64.
240    ciphertext_b64 = edq.util.encoding.to_base64(ciphertext_bytes)
241    iv_b64 = edq.util.encoding.to_base64(iv_bytes)
242    salt_b64 = edq.util.encoding.to_base64(salt_bytes)
243
244    return (ciphertext_b64, iv_b64, salt_b64)

Perform AES256-CBC encryption. Returns the base64 encoding of the ciphertext, iv, and salt. If the IV and/or salt are passed in, the same values will be passed back.

def aes256_decrypt( key: str, iv_b64: str, salt_b64: str, ciphertext_b64: str, encoding: str = 'utf-8') -> str: View Source

246def aes256_decrypt(
247        key: str,
248        iv_b64: str,
249        salt_b64: str,
250        ciphertext_b64: str,
251        encoding: str = edq.util.constants.DEFAULT_ENCODING,
252        ) -> str:
253    """
254    Perform AES256-CBC decryption.
255    """
256
257    # Get the input data as bytes.
258    iv_bytes = edq.util.encoding.from_base64(iv_b64, encoding = encoding)
259    salt_bytes = edq.util.encoding.from_base64(salt_b64, encoding = encoding)
260    ciphertext_bytes = edq.util.encoding.from_base64(ciphertext_b64, encoding = encoding)
261
262    # Derive fixed-sized keys from the key (32 bytes) and IV (16 bytes).
263    aes_key = _derive_aes_key(key.encode(encoding), salt_bytes, 32)
264    aes_iv = _derive_aes_key(iv_bytes, salt_bytes, AES_BLOCK_SIZE_BYTES)
265
266    # Decrypt the data.
267    cipher = cryptography.hazmat.primitives.ciphers.Cipher(
268        cryptography.hazmat.primitives.ciphers.algorithms.AES(aes_key),
269        cryptography.hazmat.primitives.ciphers.modes.CBC(aes_iv),
270    )
271    decryptor = cipher.decryptor()
272    cleartext_bytes_padded = decryptor.update(ciphertext_bytes) + decryptor.finalize()
273
274    # Unpad and decode the data.
275    unpadder = cryptography.hazmat.primitives.padding.PKCS7(8 * AES_BLOCK_SIZE_BYTES).unpadder()
276    try:
277        cleartext_bytes = unpadder.update(cleartext_bytes_padded) + unpadder.finalize()
278    except ValueError as ex:
279        _logger.debug("Decryption failed.", exc_info = ex)
280        raise edq.core.errors.UtilsError("Decryption failed. Do you have the correct key?")  # pylint: disable=raise-missing-from
281
282    cleartext = cleartext_bytes.decode(encoding)
283
284    return cleartext

Perform AES256-CBC decryption.