Coverage for documents/management/commands/mixins.py: 0%

1import base64 

2import os 

3from typing import TypedDict 

4 

5from cryptography.fernet import Fernet 

6from cryptography.hazmat.primitives import hashes 

7from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC 

8from django.core.management import CommandError 

9 

10from documents.settings import EXPORTER_CRYPTO_ALGO_NAME 

11from documents.settings import EXPORTER_CRYPTO_KEY_ITERATIONS_NAME 

12from documents.settings import EXPORTER_CRYPTO_KEY_SIZE_NAME 

13from documents.settings import EXPORTER_CRYPTO_SALT_NAME 

14from documents.settings import EXPORTER_CRYPTO_SETTINGS_NAME 

15 

16 

17class CryptFields(TypedDict): 

18 exporter_key: str 

19 model_name: str 

20 fields: list[str] 

21 

22 

23class CryptMixin: 

24 """ 

25 Fully based on: 

26 https://cryptography.io/en/latest/fernet/#using-passwords-with-fernet 

27 

28 To encrypt: 

29 1. Call setup_crypto providing the user provided passphrase 

30 2. Call encrypt_string with a value 

31 3. Store the returned hexadecimal representation of the value 

32 

33 To decrypt: 

34 1. Load the required parameters: 

35 a. key iterations 

36 b. key size 

37 c. key algorithm 

38 2. Call setup_crypto providing the user provided passphrase and stored salt 

39 3. Call decrypt_string with a value 

40 4. Use the returned value 

41 

42 """ 

43 

44 # This matches to Django's default for now 

45 # https://github.com/django/django/blob/adae61942/django/contrib/auth/hashers.py#L315 

46 

47 # Set the defaults to be used during export 

48 # During import, these are overridden from the loaded values to ensure decryption is possible 

49 key_iterations = 1_000_000 

50 salt_size = 16 

51 key_size = 32 

52 kdf_algorithm = "pbkdf2_sha256" 

53 

54 CRYPT_FIELDS: list[CryptFields] = [ 

55 { 

56 "exporter_key": "mail_accounts", 

57 "model_name": "paperless_mail.mailaccount", 

58 "fields": [ 

59 "password", 

60 "refresh_token", 

61 ], 

62 }, 

63 { 

64 "exporter_key": "social_tokens", 

65 "model_name": "socialaccount.socialtoken", 

66 "fields": [ 

67 "token", 

68 "token_secret", 

69 ], 

70 }, 

71 ] 

72 # O(1) lookup for per-record encryption; derived from CRYPT_FIELDS at class definition time 

73 CRYPT_FIELDS_BY_MODEL: dict[str, list[str]] = { 

74 cfg["model_name"]: cfg["fields"] for cfg in CRYPT_FIELDS 

75 } 

76 

77 def get_crypt_params(self) -> dict[str, dict[str, str | int]]: 

78 return { 

79 EXPORTER_CRYPTO_SETTINGS_NAME: { 

80 EXPORTER_CRYPTO_ALGO_NAME: self.kdf_algorithm, 

81 EXPORTER_CRYPTO_KEY_ITERATIONS_NAME: self.key_iterations, 

82 EXPORTER_CRYPTO_KEY_SIZE_NAME: self.key_size, 

83 EXPORTER_CRYPTO_SALT_NAME: self.salt, 

84 }, 

85 } 

86 

87 def load_crypt_params(self, metadata: dict) -> None: 

88 # Load up the values for setting up decryption 

89 self.kdf_algorithm: str = metadata[EXPORTER_CRYPTO_SETTINGS_NAME][ 

90 EXPORTER_CRYPTO_ALGO_NAME 

91 ] 

92 self.key_iterations: int = metadata[EXPORTER_CRYPTO_SETTINGS_NAME][ 

93 EXPORTER_CRYPTO_KEY_ITERATIONS_NAME 

94 ] 

95 self.key_size: int = metadata[EXPORTER_CRYPTO_SETTINGS_NAME][ 

96 EXPORTER_CRYPTO_KEY_SIZE_NAME 

97 ] 

98 self.salt: str = metadata[EXPORTER_CRYPTO_SETTINGS_NAME][ 

99 EXPORTER_CRYPTO_SALT_NAME 

100 ] 

101 

102 def setup_crypto(self, *, passphrase: str, salt: str | None = None) -> None: 

103 """ 

104 Constructs a class for encryption or decryption using the specified passphrase and salt 

105 

106 Salt is assumed to be a hexadecimal representation of a cryptographically secure random byte string. 

107 If not provided, it will be derived from the system secure random 

108 """ 

109 self.salt = salt or os.urandom(self.salt_size).hex() 

110 

111 # Derive the KDF based on loaded settings 

112 if self.kdf_algorithm == "pbkdf2_sha256": 

113 kdf = PBKDF2HMAC( 

114 algorithm=hashes.SHA256(), 

115 length=self.key_size, 

116 salt=bytes.fromhex(self.salt), 

117 iterations=self.key_iterations, 

118 ) 

119 else: # pragma: no cover 

120 raise CommandError( 

121 f"{self.kdf_algorithm} is an unknown key derivation function", 

122 ) 

123 

124 key = base64.urlsafe_b64encode(kdf.derive(passphrase.encode("utf-8"))) 

125 

126 self.fernet = Fernet(key) 

127 

128 def encrypt_string(self, *, value: str) -> str: 

129 """ 

130 Given a string value, encrypts it and returns the hexadecimal representation of the encrypted token 

131 

132 """ 

133 return self.fernet.encrypt(value.encode("utf-8")).hex() 

134 

135 def decrypt_string(self, *, value: str) -> str: 

136 """ 

137 Given a string value, decrypts it and returns the original value of the field 

138 """ 

139 return self.fernet.decrypt(bytes.fromhex(value)).decode("utf-8")