-
Notifications
You must be signed in to change notification settings - Fork 8
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'release/0.4.0' into main
- Loading branch information
Showing
63 changed files
with
1,764 additions
and
1,917 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,25 +1,7 @@ | ||
| from django.apps import apps as django_apps | ||
| from django.conf import settings | ||
| from django.core.exceptions import ImproperlyConfigured | ||
|
|
||
|
|
||
| def get_crypt_model(): | ||
| """ | ||
| Return the Crypt model that is active in this project. | ||
| """ | ||
| try: | ||
| DJANGO_CRYPTO_FIELDS_MODEL = settings.DJANGO_CRYPTO_FIELDS_MODEL | ||
| except AttributeError: | ||
| DJANGO_CRYPTO_FIELDS_MODEL = "django_crypto_fields.crypt" | ||
|
|
||
| try: | ||
| return django_apps.get_model(DJANGO_CRYPTO_FIELDS_MODEL, require_ready=False) | ||
| except ValueError: | ||
| raise ImproperlyConfigured( | ||
| "DJANGO_CRYPTO_FIELDS_MODEL must be of the form 'app_label.model_name'" | ||
| ) | ||
| except LookupError: | ||
| raise ImproperlyConfigured( | ||
| f"DJANGO_CRYPTO_FIELDS_MODEL refers to model {DJANGO_CRYPTO_FIELDS_MODEL} " | ||
| "that has not been installed" | ||
| ) | ||
| # import sys | ||
| # | ||
| # from .keys import encryption_keys | ||
| # | ||
| # sys.stdout.write(f"Loading encryption keys ...\n") | ||
| # encryption_keys.initialize() | ||
| # sys.stdout.write(f"Done loading encryption keys.\n") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,113 +1,33 @@ | ||
| import os | ||
| import sys | ||
| from tempfile import mkdtemp | ||
|
|
||
| from django.apps import AppConfig as DjangoAppConfig | ||
| from django.conf import settings | ||
| from django.core.checks import register | ||
| from django.core.management.color import color_style | ||
|
|
||
| from .key_creator import KeyCreator | ||
| from .key_files import KeyFiles | ||
| from .key_path import KeyPath | ||
| from .keys import Keys | ||
| from .persist_key_path import get_last_key_path | ||
| from .system_checks import aes_mode_check, encryption_keys_check, key_path_check | ||
|
|
||
|
|
||
| class DjangoCryptoFieldsError(Exception): | ||
| pass | ||
|
|
||
|
|
||
| class DjangoCryptoFieldsKeysDoNotExist(Exception): | ||
| pass | ||
|
|
||
|
|
||
| style = color_style() | ||
|
|
||
|
|
||
| class AppConfig(DjangoAppConfig): | ||
| name: str = "django_crypto_fields" | ||
| verbose_name: str = "Data Encryption" | ||
| _keys = None | ||
| _key_path_validated = None | ||
| verbose_name: str = "django-crypto-fields" | ||
| app_label: str = "django_crypto_fields" | ||
| last_key_path_filename: str = "django_crypto_fields" | ||
| key_reference_model: str = "django_crypto_fields.keyreference" | ||
| # change if using more than one database and not 'default'. | ||
| crypt_model_using: str = "default" | ||
|
|
||
| def __init__(self, app_label: str, model_name: str): | ||
| """Placed here instead of `ready()`. For models to | ||
| load correctly that use field classes from this module the keys | ||
| need to be loaded before models. | ||
| """ | ||
| self.temp_path = mkdtemp() | ||
|
|
||
| path = None | ||
| if getattr(settings, "DJANGO_CRYPTO_FIELDS_TEMP_PATH", "test" in sys.argv): | ||
| path = self.temp_path | ||
| def import_models(self): | ||
| from .keys import encryption_keys # noqa | ||
|
|
||
| self._key_path = KeyPath(path=path) | ||
| self.key_files = None | ||
| self.last_key_path = get_last_key_path(self.last_key_path_filename) | ||
|
|
||
| sys.stdout.write(f"Loading {self.verbose_name} (init)...\n") | ||
|
|
||
| self.key_files = KeyFiles(key_path=self.key_path) | ||
| if not self._keys and not self.key_files.key_files_exist: | ||
| if self.auto_create_keys: | ||
| if not os.access(self.key_path.path, os.W_OK): | ||
| raise DjangoCryptoFieldsError( | ||
| "Cannot auto-create encryption keys. Folder is not writeable." | ||
| f"Got {self.key_path}" | ||
| ) | ||
| sys.stdout.write( | ||
| style.SUCCESS(f" * settings.AUTO_CREATE_KEYS={self.auto_create_keys}.\n") | ||
| ) | ||
| key_creator = KeyCreator(key_files=self.key_files, verbose_mode=True) | ||
| key_creator.create_keys() | ||
| self._keys = Keys(key_path=self.key_path) | ||
| self._keys.load_keys() | ||
| else: | ||
| raise DjangoCryptoFieldsKeysDoNotExist( | ||
| f"Failed to find any encryption keys in path {self.key_path}. " | ||
| "If this is your first time loading " | ||
| "the project, set settings.AUTO_CREATE_KEYS=True and restart. " | ||
| "Make sure the folder is writeable." | ||
| ) | ||
| else: | ||
| self._keys = Keys(key_path=self.key_path) | ||
| self._keys.load_keys() | ||
|
|
||
| super().__init__(app_label, model_name) | ||
| sys.stdout.write(f" Done loading {self.verbose_name} (init)...\n") | ||
| return super().import_models() | ||
|
|
||
| def ready(self): | ||
| style = color_style() | ||
| path = KeyPath().path | ||
| sys.stdout.write(f"Loading {self.verbose_name} ...\n") | ||
| if "test" not in sys.argv: | ||
| register(key_path_check)(["django_crypto_fields"]) | ||
| register(encryption_keys_check)(["django_crypto_fields"]) | ||
| register(aes_mode_check) | ||
| sys.stdout.write(f" * found encryption keys in {self.key_path}.\n") | ||
| sys.stdout.write(f" * using model {self.app_label}.crypt.\n") | ||
| sys.stdout.write(f" * Keys are in folder {path}\n") | ||
| if os.access(path, os.W_OK): | ||
| sys.stdout.write( | ||
| style.WARNING(" * Remember to make folder READ-ONLY in production\n") | ||
| ) | ||
| sys.stdout.write( | ||
| style.WARNING(" * Remember to keep a backup of your encryption keys\n") | ||
| ) | ||
| sys.stdout.write(f" Done loading {self.verbose_name}.\n") | ||
|
|
||
| @property | ||
| def encryption_keys(self): | ||
| return self._keys | ||
|
|
||
| @property | ||
| def auto_create_keys(self): | ||
| try: | ||
| auto_create_keys = settings.AUTO_CREATE_KEYS | ||
| except AttributeError: | ||
| auto_create_keys = None | ||
| if "test" in sys.argv: | ||
| if auto_create_keys is None: | ||
| auto_create_keys = True | ||
| return auto_create_keys | ||
|
|
||
| @property | ||
| def key_path(self): | ||
| return self._key_path |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,4 @@ | ||
| from .cipher import Cipher | ||
| from .cipher_parser import CipherParser | ||
|
|
||
| __all__ = ["Cipher", "CipherParser"] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| from __future__ import annotations | ||
|
|
||
| from typing import Callable | ||
|
|
||
| from ..constants import CIPHER_PREFIX, HASH_PREFIX | ||
| from ..utils import make_hash, safe_encode_utf8 | ||
|
|
||
| __all__ = ["Cipher"] | ||
|
|
||
|
|
||
| class Cipher: | ||
| """A class that given a value builds a cipher of the format | ||
| hash_prefix + hashed_value + cipher_prefix + secret. | ||
| . | ||
| For example: | ||
| enc1:::234234ed234a24enc2::\x0e\xb9\xae\x13s\x8d | ||
| \xe7O\xbb\r\x99. | ||
| The secret is encrypted using the passed `encrypt` callable. | ||
| """ | ||
|
|
||
| def __init__( | ||
| self, | ||
| value: str | bytes, | ||
| salt_key: bytes, | ||
| encrypt: Callable[[bytes], bytes] | None = None, | ||
| ): | ||
| encoded_value = safe_encode_utf8(value) | ||
| self.hash_prefix = b"" | ||
| self.hashed_value = b"" | ||
| self.cipher_prefix = b"" | ||
| self.secret = b"" | ||
| if salt_key: | ||
| self.hash_prefix: bytes = safe_encode_utf8(HASH_PREFIX) | ||
| self.hashed_value: bytes = make_hash(encoded_value, salt_key) | ||
| if encrypt: | ||
| self.secret = encrypt(encoded_value) | ||
| self.cipher_prefix: bytes = safe_encode_utf8(CIPHER_PREFIX) | ||
|
|
||
| @property | ||
| def cipher(self) -> bytes: | ||
| return self.hash_with_prefix + self.secret_with_prefix | ||
|
|
||
| @property | ||
| def hash_with_prefix(self) -> bytes: | ||
| return self.hash_prefix + self.hashed_value | ||
|
|
||
| @property | ||
| def secret_with_prefix(self) -> bytes: | ||
| return self.cipher_prefix + self.secret |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| from __future__ import annotations | ||
|
|
||
| from ..constants import CIPHER_PREFIX, HASH_PREFIX | ||
| from ..exceptions import MalformedCiphertextError | ||
| from ..utils import make_hash, safe_encode_utf8 | ||
|
|
||
| __all__ = ["CipherParser"] | ||
|
|
||
|
|
||
| class CipherParser: | ||
| def __init__(self, cipher: bytes, salt_key: bytes | None = None): | ||
| self._cipher_prefix = None | ||
| self._hash_prefix = None | ||
| self._hashed_value = None | ||
| self._secret = None | ||
| self.cipher = safe_encode_utf8(cipher) | ||
| self.salt_key = salt_key | ||
| self.validate_hashed_value() | ||
| self.validate_secret() | ||
|
|
||
| @property | ||
| def hash_prefix(self) -> bytes | None: | ||
| if self.cipher: | ||
| hash_prefix = safe_encode_utf8(HASH_PREFIX) | ||
| self._hash_prefix = hash_prefix if self.cipher.startswith(hash_prefix) else None | ||
| return self._hash_prefix | ||
|
|
||
| @property | ||
| def cipher_prefix(self) -> bytes | None: | ||
| if self.cipher: | ||
| cipher_prefix = safe_encode_utf8(CIPHER_PREFIX) | ||
| self._cipher_prefix = cipher_prefix if cipher_prefix in self.cipher else None | ||
| return self._cipher_prefix | ||
|
|
||
| @property | ||
| def hashed_value(self) -> bytes | None: | ||
| if self.cipher and self.cipher.startswith(self.hash_prefix): | ||
| self._hashed_value = self.cipher.split(self.hash_prefix)[1].split( | ||
| self.cipher_prefix | ||
| )[0] | ||
| return self._hashed_value | ||
|
|
||
| @property | ||
| def secret(self) -> bytes | None: | ||
| if self.cipher and safe_encode_utf8(CIPHER_PREFIX) in self.cipher: | ||
| self._secret = self.cipher.split(self.cipher_prefix)[1] | ||
| return self._secret | ||
|
|
||
| def validate_hashed_value(self) -> None: | ||
| if self.hash_prefix and not self.hashed_value: | ||
| raise MalformedCiphertextError("Invalid hashed_value. Got None.") | ||
| elif self.salt_key and len(self.hashed_value) != len(make_hash("Foo", self.salt_key)): | ||
| raise MalformedCiphertextError("Invalid hashed_value. Incorrect size.") | ||
|
|
||
| def validate_secret(self) -> None: | ||
| if self.cipher_prefix and not self.secret: | ||
| raise MalformedCiphertextError("Invalid secret. Got None.") |
Oops, something went wrong.