Add support for switching cipher methods (#156)

* Add support for switching cipher methods through PNConfiguration

* Validation of cipher methods

* Default - CBC, fallback - None

* Add fallback to file crypto

* PubNub SDK 7.2.0 release.

---------

Co-authored-by: PubNub Release Bot <[email protected]>

.pubnub.yml

@@ -1,5 +1,5 @@
 name: python
-version: 7.1.0
+version: 7.2.0
 schema: 1
 scm: github.com/pubnub/python
 sdks:
@@ -18,7 +18,7 @@ sdks:
         distributions:
           - distribution-type: library
             distribution-repository: package
-            package-name: pubnub-7.1.0
+            package-name: pubnub-7.2.0
             location: https://pypi.org/project/pubnub/
             supported-platforms:
               supported-operating-systems:
@@ -97,8 +97,8 @@ sdks:
           -
             distribution-type: library
             distribution-repository: git release
-            package-name: pubnub-7.1.0
-            location: https://github.com/pubnub/python/releases/download/7.1.0/pubnub-7.1.0.tar.gz
+            package-name: pubnub-7.2.0
+            location: https://github.com/pubnub/python/releases/download/7.2.0/pubnub-7.2.0.tar.gz
             supported-platforms:
               supported-operating-systems:
                 Linux:
@@ -169,6 +169,11 @@ sdks:
                 license-url: https://github.com/aio-libs/aiohttp/blob/master/LICENSE.txt
                 is-required: Required
 changelog:
+  - date: 2023-07-06
+    version: 7.2.0
+    changes:
+      - type: feature
+        text: "Introduced option to select ciphering method for encoding messages and files. The default behavior is unchanged. More can be read [in this comment](https://github.com/pubnub/python/pull/156#issuecomment-1623307799)."
   - date: 2023-01-17
     version: 7.1.0
     changes:

CHANGELOG.md

@@ -1,3 +1,9 @@
+## 7.2.0
+July 06 2023
+
+#### Added
+- Introduced option to select ciphering method for encoding messages and files. The default behavior is unchanged. More can be read [in this comment](https://github.com/pubnub/python/pull/156#issuecomment-1623307799).
+
 ## 7.1.0
 January 17 2023
 

examples/crypto.py

@@ -0,0 +1,47 @@
+from Cryptodome.Cipher import AES
+from os import getenv
+from pubnub.pnconfiguration import PNConfiguration
+from pubnub.pubnub import PubNub
+from time import sleep
+
+channel = 'cipher_algorithm_experiment'
+
+
+def PNFactory(cipher_mode=AES.MODE_GCM, fallback_cipher_mode=AES.MODE_CBC) -> PubNub:
+    config = config = PNConfiguration()
+    config.publish_key = getenv('PN_KEY_PUBLISH')
+    config.subscribe_key = getenv('PN_KEY_SUBSCRIBE')
+    config.secret_key = getenv('PN_KEY_SECRET')
+    config.cipher_key = getenv('PN_KEY_CIPHER')
+    config.user_id = 'experiment'
+    config.cipher_mode = cipher_mode
+    config.fallback_cipher_mode = fallback_cipher_mode
+
+    return PubNub(config)
+
+
+# let's build history with legacy AES.CBC
+pn = PNFactory(cipher_mode=AES.MODE_CBC, fallback_cipher_mode=None)
+pn.publish().channel(channel).message('message encrypted with CBC').sync()
+pn.publish().channel(channel).message('message encrypted with CBC').sync()
+
+# now with upgraded config
+pn = PNFactory(cipher_mode=AES.MODE_GCM, fallback_cipher_mode=AES.MODE_CBC)
+pn.publish().channel(channel).message('message encrypted with GCM').sync()
+pn.publish().channel(channel).message('message encrypted with GCM').sync()
+
+# give some time to store messages
+sleep(3)
+
+# after upgrade decoding with GCM and fallback CBC
+pn = PNFactory(cipher_mode=AES.MODE_GCM, fallback_cipher_mode=AES.MODE_CBC)
+messages = pn.history().channel(channel).sync()
+print([message.entry for message in messages.result.messages])
+
+# before upgrade decoding with CBC and without fallback
+pn = PNFactory(cipher_mode=AES.MODE_CBC, fallback_cipher_mode=None)
+try:
+    messages = pn.history().channel(channel).sync()
+    print([message.entry for message in messages.result.messages])
+except UnicodeDecodeError:
+    print('Unable to decode - Exception has been thrown')

pubnub/crypto.py

@@ -3,7 +3,7 @@
 import random
 from base64 import decodebytes, encodebytes
 
-from .crypto_core import PubNubCrypto
+from pubnub.crypto_core import PubNubCrypto
 from Cryptodome.Cipher import AES
 from Cryptodome.Util.Padding import pad, unpad
 
@@ -12,14 +12,19 @@
 
 
 class PubNubCryptodome(PubNubCrypto):
+    mode = AES.MODE_CBC
+    fallback_mode = None
+
     def __init__(self, pubnub_config):
         self.pubnub_configuration = pubnub_config
+        self.mode = pubnub_config.cipher_mode
+        self.fallback_mode = pubnub_config.fallback_cipher_mode
 
     def encrypt(self, key, msg, use_random_iv=False):
         secret = self.get_secret(key)
         initialization_vector = self.get_initialization_vector(use_random_iv)
 
-        cipher = AES.new(bytes(secret[0:32], 'utf-8'), AES.MODE_CBC, bytes(initialization_vector, 'utf-8'))
+        cipher = AES.new(bytes(secret[0:32], 'utf-8'), self.mode, bytes(initialization_vector, 'utf-8'))
         encrypted_message = cipher.encrypt(self.pad(msg.encode('utf-8')))
         msg_with_iv = self.append_random_iv(encrypted_message, use_random_iv, bytes(initialization_vector, "utf-8"))
 
@@ -30,8 +35,15 @@ def decrypt(self, key, msg, use_random_iv=False):
 
         decoded_message = decodebytes(msg.encode("utf-8"))
         initialization_vector, extracted_message = self.extract_random_iv(decoded_message, use_random_iv)
-        cipher = AES.new(bytes(secret[0:32], "utf-8"), AES.MODE_CBC, initialization_vector)
-        plain = self.depad((cipher.decrypt(extracted_message)).decode('utf-8'))
+        cipher = AES.new(bytes(secret[0:32], "utf-8"), self.mode, initialization_vector)
+        try:
+            plain = self.depad((cipher.decrypt(extracted_message)).decode('utf-8'))
+        except UnicodeDecodeError as e:
+            if not self.fallback_mode:
+                raise e
+
+            cipher = AES.new(bytes(secret[0:32], "utf-8"), self.fallback_mode, initialization_vector)
+            plain = self.depad((cipher.decrypt(extracted_message)).decode('utf-8'))
 
         try:
             return json.loads(plain)
@@ -71,7 +83,7 @@ class PubNubFileCrypto(PubNubCryptodome):
     def encrypt(self, key, file):
         secret = self.get_secret(key)
         initialization_vector = self.get_initialization_vector(use_random_iv=True)
-        cipher = AES.new(bytes(secret[0:32], "utf-8"), AES.MODE_CBC, bytes(initialization_vector, 'utf-8'))
+        cipher = AES.new(bytes(secret[0:32], "utf-8"), self.mode, bytes(initialization_vector, 'utf-8'))
         initialization_vector = bytes(initialization_vector, 'utf-8')
 
         return self.append_random_iv(
@@ -83,6 +95,11 @@ def encrypt(self, key, file):
     def decrypt(self, key, file):
         secret = self.get_secret(key)
         initialization_vector, extracted_file = self.extract_random_iv(file, use_random_iv=True)
-        cipher = AES.new(bytes(secret[0:32], "utf-8"), AES.MODE_CBC, initialization_vector)
+        try:
+            cipher = AES.new(bytes(secret[0:32], "utf-8"), self.mode, initialization_vector)
+            result = unpad(cipher.decrypt(extracted_file), 16)
+        except ValueError:
+            cipher = AES.new(bytes(secret[0:32], "utf-8"), self.fallback_mode, initialization_vector)
+            result = unpad(cipher.decrypt(extracted_file), 16)
 
-        return unpad(cipher.decrypt(extracted_file), 16)
+        return result

pubnub/pnconfiguration.py

@@ -1,9 +1,12 @@
-from .enums import PNHeartbeatNotificationOptions, PNReconnectionPolicy
+from Cryptodome.Cipher import AES
+from pubnub.enums import PNHeartbeatNotificationOptions, PNReconnectionPolicy
+from pubnub.exceptions import PubNubException
 
 
 class PNConfiguration(object):
     DEFAULT_PRESENCE_TIMEOUT = 300
     DEFAULT_HEARTBEAT_INTERVAL = 280
+    ALLOWED_AES_MODES = [AES.MODE_CBC, AES.MODE_GCM]
 
     def __init__(self):
         # TODO: add validation
@@ -17,6 +20,8 @@ def __init__(self):
         self.publish_key = None
         self.secret_key = None
         self.cipher_key = None
+        self._cipher_mode = AES.MODE_CBC
+        self._fallback_cipher_mode = None
         self.auth_key = None
         self.filter_expression = None
         self.enable_subscribe = True
@@ -61,6 +66,30 @@ def set_presence_timeout_with_custom_interval(self, timeout, interval):
     def set_presence_timeout(self, timeout):
         self.set_presence_timeout_with_custom_interval(timeout, (timeout / 2) - 1)
 
+    @property
+    def cipher_mode(self):
+        return self._cipher_mode
+
+    @cipher_mode.setter
+    def cipher_mode(self, cipher_mode):
+        if cipher_mode not in self.ALLOWED_AES_MODES:
+            raise PubNubException('Cipher mode not supported')
+        if cipher_mode is not self._cipher_mode:
+            self._cipher_mode = cipher_mode
+            self.crypto_instance = None
+
+    @property
+    def fallback_cipher_mode(self):
+        return self._fallback_cipher_mode
+
+    @fallback_cipher_mode.setter
+    def fallback_cipher_mode(self, fallback_cipher_mode):
+        if fallback_cipher_mode not in self.ALLOWED_AES_MODES:
+            raise PubNubException('Cipher mode not supported')
+        if fallback_cipher_mode is not self._fallback_cipher_mode:
+            self._fallback_cipher_mode = fallback_cipher_mode
+            self.crypto_instance = None
+
     @property
     def crypto(self):
         if self.crypto_instance is None:

pubnub/pubnub_core.py

@@ -83,7 +83,7 @@
 
 class PubNubCore:
     """A base class for PubNub Python API implementations"""
-    SDK_VERSION = "7.1.0"
+    SDK_VERSION = "7.2.0"
     SDK_NAME = "PubNub-Python"
 
     TIMESTAMP_DIVIDER = 1000

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name='pubnub',
-    version='7.1.0',
+    version='7.2.0',
     description='PubNub Real-time push service in the cloud',
     author='PubNub',
     author_email='[email protected]',