Merge pull request #35642 from dimagi/jt/cbc-password-read-write

Geoconfig and API settings models: AES CBC encryption read and write

corehq/apps/geospatial/const.py

@@ -1,8 +1,6 @@
 
 GPS_POINT_CASE_PROPERTY = 'gps_point'
 
-ALGO_AES = 'aes'
-
 # Max number of cases per geohash
 MAX_GEOHASH_DOC_COUNT = 1_000
 

corehq/apps/geospatial/models.py

@@ -5,13 +5,17 @@
 
 from corehq.apps.geospatial.const import (
     GPS_POINT_CASE_PROPERTY,
-    ALGO_AES,
     TRAVEL_MODE_WALKING,
     TRAVEL_MODE_CYCLING,
     TRAVEL_MODE_DRIVING,
 )
 from corehq.apps.geospatial.routing_solvers import pulp
-from corehq.motech.utils import b64_aes_encrypt, b64_aes_decrypt
+from corehq.motech.const import ALGO_AES, ALGO_AES_CBC
+from corehq.motech.utils import (
+    b64_aes_decrypt,
+    b64_aes_cbc_decrypt,
+    b64_aes_cbc_encrypt,
+)
 
 
 class GeoPolygon(models.Model):
@@ -112,9 +116,13 @@ def disbursement_solver(self):
 
     @property
     def plaintext_api_token(self):
-        if self.api_token and self.api_token.startswith(f'${ALGO_AES}$'):
-            ciphertext = self.api_token.split('$', 2)[2]
-            return b64_aes_decrypt(ciphertext)
+        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)
         return self.api_token
 
     @plaintext_api_token.setter
@@ -124,9 +132,9 @@ def plaintext_api_token(self, value):
         else:
             assert isinstance(value, str), "Only string values allowed for api token"
 
-            if value and not value.startswith(f'${ALGO_AES}$'):
-                ciphertext = b64_aes_encrypt(value)
-                self.api_token = f'${ALGO_AES}${ciphertext}'
+            if value and not value.startswith(f'${ALGO_AES_CBC}$'):
+                ciphertext = b64_aes_cbc_encrypt(value)
+                self.api_token = f'${ALGO_AES_CBC}${ciphertext}'
             else:
                 raise Exception("Unexpected value set for plaintext api token")
 

corehq/apps/geospatial/tests/test_models.py

@@ -2,9 +2,10 @@
 
 from django.test import TestCase
 
-from ..const import GPS_POINT_CASE_PROPERTY, ALGO_AES
+from ..const import GPS_POINT_CASE_PROPERTY
 from ..models import GeoConfig
 from ..utils import get_geo_case_property
+from corehq.motech.const import ALGO_AES_CBC
 
 
 class TestGeoConfig(TestCase):
@@ -25,7 +26,7 @@ def test_geo_config_api_token(self):
         with self.get_geo_config() as config:
             config.plaintext_api_token = '1234'
             self.assertEqual(config.plaintext_api_token, '1234')
-            self.assertTrue(config.api_token.startswith(f"${ALGO_AES}$"))
+            self.assertTrue(config.api_token.startswith(f"${ALGO_AES_CBC}$"))
 
             config.plaintext_api_token = None
             self.assertEqual(config.plaintext_api_token, None)
@@ -48,7 +49,7 @@ def test_geo_config_api_token_cannot_be_empty(self):
     def test_geo_config_api_token_cannot_start_with_encryption_str(self):
         with self.assertRaises(Exception) as context:
             with self.get_geo_config() as config:
-                config.plaintext_api_token = f"${ALGO_AES}$1234"
+                config.plaintext_api_token = f"${ALGO_AES_CBC}$1234"
 
         self.assertEqual(str(context.exception), "Unexpected value set for plaintext api token")
 

corehq/motech/models.py

@@ -24,6 +24,7 @@
 )
 from corehq.motech.const import (
     ALGO_AES,
+    ALGO_AES_CBC,
     AUTH_TYPES,
     BASIC_AUTH,
     BEARER_AUTH,
@@ -34,7 +35,11 @@
     OAUTH2_PWD,
     PASSWORD_PLACEHOLDER, APIKEY_AUTH,
 )
-from corehq.motech.utils import b64_aes_decrypt, b64_aes_encrypt
+from corehq.motech.utils import (
+    b64_aes_decrypt,
+    b64_aes_cbc_decrypt,
+    b64_aes_cbc_encrypt,
+)
 from corehq.util import as_json_text, as_text
 
 
@@ -126,35 +131,47 @@ def repeaters(self):
 
     @property
     def plaintext_password(self):
-        if self.password.startswith(f'${ALGO_AES}$'):
+        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
     def plaintext_password(self, plaintext):
         if plaintext != PASSWORD_PLACEHOLDER:
-            ciphertext = b64_aes_encrypt(plaintext)
-            self.password = f'${ALGO_AES}${ciphertext}'
+            ciphertext = b64_aes_cbc_encrypt(plaintext)
+            self.password = f'${ALGO_AES_CBC}${ciphertext}'
 
     @property
     def plaintext_client_secret(self):
-        if self.client_secret.startswith(f'${ALGO_AES}$'):
+        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
     def plaintext_client_secret(self, plaintext):
         if plaintext != PASSWORD_PLACEHOLDER:
-            ciphertext = b64_aes_encrypt(plaintext)
-            self.client_secret = f'${ALGO_AES}${ciphertext}'
+            ciphertext = b64_aes_cbc_encrypt(plaintext)
+            self.client_secret = f'${ALGO_AES_CBC}${ciphertext}'
 
     @property
     def last_token(self) -> Optional[dict]:
         if self.last_token_aes:
-            plaintext = b64_aes_decrypt(self.last_token_aes)
-            return json.loads(plaintext)
+            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)
         return None
 
     @last_token.setter
@@ -163,7 +180,8 @@ def last_token(self, token: Optional[dict]):
             self.last_token_aes = ''
         else:
             plaintext = json.dumps(token)
-            self.last_token_aes = b64_aes_encrypt(plaintext)
+            ciphertext = b64_aes_cbc_encrypt(plaintext)
+            self.last_token_aes = f'${ALGO_AES_CBC}${ciphertext}'
 
     @property
     def notify_addresses(self):

corehq/motech/tests/test_auth.py

@@ -1,7 +1,7 @@
 from django.test import TestCase
 
 from corehq.motech.auth import BasicAuthManager
-from corehq.motech.const import ALGO_AES, BASIC_AUTH
+from corehq.motech.const import ALGO_AES_CBC, BASIC_AUTH
 from corehq.motech.models import ConnectionSettings
 
 
@@ -30,4 +30,4 @@ def test_connection_settings_auth_manager(self):
         self.assertEqual(auth_manager.username, self.username)
         self.assertEqual(auth_manager.password, self.password)
         self.assertNotEqual(auth_manager.password, self.connx.password)
-        self.assertTrue(self.connx.password.startswith(f'${ALGO_AES}$'))
+        self.assertTrue(self.connx.password.startswith(f'${ALGO_AES_CBC}$'))

corehq/motech/tests/test_models.py

@@ -8,13 +8,14 @@
 from unittest.mock import ANY, Mock, patch
 
 from corehq.apps.hqwebapp.templatetags.hq_shared_tags import pp_json
-from corehq.motech.const import ALGO_AES, PASSWORD_PLACEHOLDER
+from corehq.motech.const import ALGO_AES_CBC, PASSWORD_PLACEHOLDER
 from corehq.motech.models import (
     ConnectionSettings,
     RequestLog,
     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/'
@@ -163,15 +164,26 @@ def test_client_secret_placeholder(self):
         cs.plaintext_client_secret = PASSWORD_PLACEHOLDER
         self.assertEqual(cs.client_secret, '')
 
+    def test_last_token_setter_none(self):
+        cs = ConnectionSettings()
+        cs.last_token = None
+        self.assertEqual(cs.last_token_aes, '')
+
     def test_password_setter(self):
         cs = ConnectionSettings()
         cs.plaintext_password = 'secret'
-        self.assertTrue(cs.password.startswith(f'${ALGO_AES}$'))
+        self.assertTrue(cs.password.startswith(f'${ALGO_AES_CBC}$'))
 
     def test_client_secret_setter(self):
         cs = ConnectionSettings()
         cs.plaintext_client_secret = 'secret'
-        self.assertTrue(cs.client_secret.startswith(f'${ALGO_AES}$'))
+        self.assertTrue(cs.client_secret.startswith(f'${ALGO_AES_CBC}$'))
+
+    def test_last_token_setter(self):
+        cs = ConnectionSettings()
+        token = {'key': 'value'}
+        cs.last_token = token
+        self.assertTrue(cs.last_token_aes.startswith(f'${ALGO_AES_CBC}$'))
 
     def test_password_getter_decrypts(self):
         cs = ConnectionSettings()
@@ -183,6 +195,19 @@ def test_client_secret_getter_decrypts(self):
         cs.plaintext_client_secret = 'secret'
         self.assertEqual(cs.plaintext_client_secret, 'secret')
 
+    def test_last_token_getter_decrypts_cbc(self):
+        cs = ConnectionSettings()
+        token = {'key': 'value'}
+        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'
@@ -193,6 +218,11 @@ def test_client_secret_getter_returns(self):
         cs.client_secret = 'secret'
         self.assertEqual(cs.plaintext_client_secret, 'secret')
 
+    def test_last_token_getter_returns(self):
+        cs = ConnectionSettings()
+        cs.last_token_aes = ''
+        self.assertIsNone(cs.last_token)
+
 
 class NotifyAddressesTests(SimpleTestCase):
 

corehq/motech/utils.py

@@ -132,6 +132,10 @@ def b64_aes_cbc_decrypt(message):
     return plaintext_bytes.decode('utf8')
 
 
+class AesEcbDecryptionError(Exception):
+    pass
+
+
 # Only needed for migration from ECB to CBC mode.
 def reencrypt_ecb_to_cbc_mode(encrypted_text, existing_prefix=None):
     """
@@ -144,8 +148,11 @@ def reencrypt_ecb_to_cbc_mode(encrypted_text, existing_prefix=None):
         ciphertext = encrypted_text[len(existing_prefix):]
     else:
         ciphertext = encrypted_text
-
-    new_ciphertext = b64_aes_cbc_encrypt(b64_aes_decrypt(ciphertext))
+    try:
+        plaintext = b64_aes_decrypt(ciphertext)
+    except UnicodeDecodeError:
+        raise AesEcbDecryptionError("Failed to decrypt the AES-ECB-encrypted text.")
+    new_ciphertext = b64_aes_cbc_encrypt(plaintext)
     return f'${ALGO_AES_CBC}${new_ciphertext}'