Merge pull request #35681 from dimagi/jt/clean-up-encryption-transition

AES CBC encryption read and write Clean Up

corehq/apps/email/models.py

@@ -1,10 +1,9 @@
 from django.db import models
 
-from corehq.motech.const import PASSWORD_PLACEHOLDER, ALGO_AES, ALGO_AES_CBC
+from corehq.motech.const import PASSWORD_PLACEHOLDER, ALGO_AES_CBC
 from corehq.motech.utils import (
     b64_aes_cbc_decrypt,
     b64_aes_cbc_encrypt,
-    b64_aes_decrypt,
 )
 
 
@@ -31,9 +30,6 @@ def plaintext_password(self):
         if self.password.startswith(f'${ALGO_AES_CBC}$'):
             ciphertext = self.password.split('$', 2)[2]
             return b64_aes_cbc_decrypt(ciphertext)
-        if self.password.startswith(f'${ALGO_AES}$'):  # This will be deleted after migration to cbc is done
-            ciphertext = self.password.split('$', 2)[2]
-            return b64_aes_decrypt(ciphertext)
         return self.password
 
     @plaintext_password.setter

corehq/apps/geospatial/models.py

@@ -10,9 +10,8 @@
     TRAVEL_MODE_DRIVING,
 )
 from corehq.apps.geospatial.routing_solvers import pulp
-from corehq.motech.const import ALGO_AES, ALGO_AES_CBC
+from corehq.motech.const import ALGO_AES_CBC
 from corehq.motech.utils import (
-    b64_aes_decrypt,
     b64_aes_cbc_decrypt,
     b64_aes_cbc_encrypt,
 )
@@ -117,12 +116,8 @@ def disbursement_solver(self):
     @property
     def plaintext_api_token(self):
         if self.api_token:
-            if self.api_token.startswith(f'${ALGO_AES}$'):  # This will be deleted after migration to cbc is done
-                ciphertext = self.api_token.split('$', 2)[2]
-                return b64_aes_decrypt(ciphertext)
-            elif self.api_token.startswith(f'${ALGO_AES_CBC}$'):
-                ciphertext = self.api_token.split('$', 2)[2]
-                return b64_aes_cbc_decrypt(ciphertext)
+            ciphertext = self.api_token.split('$', 2)[2]
+            return b64_aes_cbc_decrypt(ciphertext)
         return self.api_token
 
     @plaintext_api_token.setter

corehq/apps/reports/models.py

@@ -21,7 +21,6 @@
 from corehq.apps.users.models import CommCareUser
 from corehq.motech.const import ALGO_AES_CBC
 from corehq.motech.utils import (
-    b64_aes_decrypt,
     b64_aes_cbc_encrypt,
     b64_aes_cbc_decrypt
 )
@@ -209,11 +208,10 @@ def __str__(self):
 
     @property
     def plaintext_secret_value(self):
-        # Conditonal check  be deleted after migration to cbc is done
-        if self.encrypted_secret_value.startswith(f'${ALGO_AES_CBC}$'):
-            ciphertext = self.encrypted_secret_value.split('$', 2)[2]
-            return b64_aes_cbc_decrypt(ciphertext)
-        return b64_aes_decrypt(self.encrypted_secret_value)  # This will be deleted after migration to cbc is done
+        if self.encrypted_secret_value == '':
+            return ''
+        ciphertext = self.encrypted_secret_value.split('$', 2)[2]
+        return b64_aes_cbc_decrypt(ciphertext)
 
     @plaintext_secret_value.setter
     def plaintext_secret_value(self, plaintext):

corehq/apps/translations/models.py

@@ -7,7 +7,6 @@
 from corehq.apps.app_manager.dbaccessors import get_app, get_app_ids_in_domain
 from corehq.motech.const import ALGO_AES_CBC
 from corehq.motech.utils import (
-    b64_aes_decrypt,
     b64_aes_cbc_decrypt,
     b64_aes_cbc_encrypt,
 )
@@ -171,10 +170,10 @@ def __str__(self):
 
     @property
     def plaintext_api_token(self):
-        if self.api_token.startswith(f'${ALGO_AES_CBC}$'):
-            ciphertext = self.api_token.split('$', 2)[2]
-            return b64_aes_cbc_decrypt(ciphertext)
-        return b64_aes_decrypt(self.api_token)
+        if self.api_token == '':
+            return ''
+        ciphertext = self.api_token.split('$', 2)[2]
+        return b64_aes_cbc_decrypt(ciphertext)
 
     @plaintext_api_token.setter
     def plaintext_api_token(self, plaintext):

corehq/motech/models.py

@@ -23,7 +23,6 @@
     ApiKeyAuthManager,
 )
 from corehq.motech.const import (
-    ALGO_AES,
     ALGO_AES_CBC,
     AUTH_TYPES,
     BASIC_AUTH,
@@ -36,7 +35,6 @@
     PASSWORD_PLACEHOLDER, APIKEY_AUTH,
 )
 from corehq.motech.utils import (
-    b64_aes_decrypt,
     b64_aes_cbc_decrypt,
     b64_aes_cbc_encrypt,
 )
@@ -134,9 +132,6 @@ def plaintext_password(self):
         if self.password.startswith(f'${ALGO_AES_CBC}$'):
             ciphertext = self.password.split('$', 2)[2]
             return b64_aes_cbc_decrypt(ciphertext)
-        elif self.password.startswith(f'${ALGO_AES}$'):  # This will be deleted after migration to cbc is done
-            ciphertext = self.password.split('$', 2)[2]
-            return b64_aes_decrypt(ciphertext)
         return self.password
 
     @plaintext_password.setter
@@ -150,9 +145,6 @@ def plaintext_client_secret(self):
         if self.client_secret.startswith(f'${ALGO_AES_CBC}$'):
             ciphertext = self.client_secret.split('$', 2)[2]
             return b64_aes_cbc_decrypt(ciphertext)
-        elif self.client_secret.startswith(f'${ALGO_AES}$'):  # This will be deleted after migration to cbc is done
-            ciphertext = self.client_secret.split('$', 2)[2]
-            return b64_aes_decrypt(ciphertext)
         return self.client_secret
 
     @plaintext_client_secret.setter
@@ -164,14 +156,9 @@ def plaintext_client_secret(self, plaintext):
     @property
     def last_token(self) -> Optional[dict]:
         if self.last_token_aes:
-            if self.last_token_aes.startswith(f'${ALGO_AES_CBC}$'):
-                ciphertext = self.last_token_aes.split('$', 2)[2]
-                plaintext = b64_aes_cbc_decrypt(ciphertext)
-                return json.loads(plaintext)
-            else:
-                # This will be deleted after migration to cbc is done
-                plaintext = b64_aes_decrypt(self.last_token_aes)
-                return json.loads(plaintext)
+            ciphertext = self.last_token_aes.split('$', 2)[2]
+            plaintext = b64_aes_cbc_decrypt(ciphertext)
+            return json.loads(plaintext)
         return None
 
     @last_token.setter

corehq/motech/openmrs/models.py

@@ -22,9 +22,8 @@
     IMPORT_FREQUENCY_DAILY,
     IMPORT_FREQUENCY_MONTHLY,
     IMPORT_FREQUENCY_WEEKLY,
-    ALGO_AES_CBC,
 )
-from corehq.motech.utils import b64_aes_decrypt, b64_aes_cbc_decrypt
+from corehq.motech.utils import b64_aes_cbc_decrypt
 from corehq.motech.openmrs.const import (
     OPENMRS_DATA_TYPE_MILLISECONDS,
     OPENMRS_DATA_TYPES,
@@ -134,10 +133,10 @@ def notify_addresses(self):
 
     @property
     def plaintext_password(self):
-        if self.password.startswith(f'${ALGO_AES_CBC}$'):
-            ciphertext = self.password.split('$', 2)[2]
-            return b64_aes_cbc_decrypt(ciphertext)
-        return b64_aes_decrypt(self.password)
+        if self.password == '':
+            return ''
+        ciphertext = self.password.split('$', 2)[2]
+        return b64_aes_cbc_decrypt(ciphertext)
 
     @memoized
     def get_timezone(self):

corehq/motech/openmrs/tests/test_tasks.py

@@ -17,14 +17,15 @@
 from corehq.apps.users.models import CommCareUser, WebUser
 from corehq.motech.auth import AuthManager
 from corehq.motech.exceptions import ConfigurationError
-from corehq.motech.const import IMPORT_FREQUENCY_MONTHLY
+from corehq.motech.const import IMPORT_FREQUENCY_MONTHLY, ALGO_AES_CBC
 from corehq.motech.openmrs.models import OpenmrsImporter
 from corehq.motech.openmrs.tasks import (
     get_case_properties,
     get_openmrs_patients,
     import_patients_with_importer,
 )
 from corehq.motech.requests import Requests
+from corehq.motech.utils import b64_aes_cbc_encrypt
 from corehq.motech.views import ConnectionSettingsListView
 from corehq.util.view_utils import absolute_reverse
 
@@ -37,7 +38,7 @@ def get_importer(column_mapping=None):
         'domain': TEST_DOMAIN,
         'server_url': 'http://www.example.com/openmrs',
         'username': 'admin',
-        'password': 'Admin123',
+        'password': f"${ALGO_AES_CBC}${b64_aes_cbc_encrypt('Admin123')}",
         'notify_addresses_str': '[email protected]',
         'location_id': '',
         'import_frequency': IMPORT_FREQUENCY_MONTHLY,
@@ -338,7 +339,7 @@ def setUp(self):
         self.send_mail_mock = self.send_mail_patcher.start()
         self.import_patcher = patch('corehq.motech.openmrs.tasks.import_patients_of_owner')
         self.import_mock = self.import_patcher.start()
-        self.decrypt_patcher = patch('corehq.motech.openmrs.models.b64_aes_decrypt')
+        self.decrypt_patcher = patch('corehq.motech.openmrs.models.b64_aes_cbc_decrypt')
         self.decrypt_patcher.start()
 
     def tearDown(self):

corehq/motech/tests/test_models.py

@@ -15,7 +15,6 @@
     RequestLogEntry,
 )
 from corehq.motech.requests import get_basic_requests
-from corehq.motech.utils import b64_aes_encrypt
 from corehq.util import as_json_text, as_text
 
 TEST_API_URL = 'http://example.com:9080/api/'
@@ -201,13 +200,6 @@ def test_last_token_getter_decrypts_cbc(self):
         cs.last_token = token
         self.assertEqual(cs.last_token, token)
 
-    def test_last_token_getter_decrypts_ecb(self):
-        cs = ConnectionSettings()
-        token = {'key': 'value'}
-        plaintext = json.dumps(token)
-        cs.last_token_aes = b64_aes_encrypt(plaintext)
-        self.assertEqual(cs.last_token, token)
-
     def test_password_getter_returns(self):
         cs = ConnectionSettings()
         cs.password = 'secret'