Merged in KRAUS-10 (pull request #14)

KRAUS-10: Updated property_encryptor.py to leverage the cryptography package instead of pycryptodome to provide PBEWithMD5AndDES encryption/decryption support

Approved-by: Aaron Gary

Author: ekoniec1

krausening-python/poetry.lock

@@ -13,13 +13,11 @@ keywords = ["properties", "configuration-management"]
 python = "3.9.13"
 javaproperties = "^0.8.1"
 cryptography = "^37.0.4"
-typing_extensions = "^4.1.1"
-pycryptodome = "^3.15.0"
 
 [tool.poetry.dev-dependencies]
 behave = "^1.2.6"
-nose = "^1.3.7"
 black = "^22.6.0"
+nose = "^1.3.7"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

krausening-python/pyproject.toml

@@ -1,42 +1,60 @@
 import base64
-import hashlib
 import re
 import os
-from Crypto.Cipher import DES
+
+from cryptography.hazmat.primitives.hashes import Hash, MD5
+from cryptography.hazmat.primitives.ciphers import Cipher
+from cryptography.hazmat.primitives.ciphers.algorithms import TripleDES
+from cryptography.hazmat.primitives.ciphers.modes import CBC
 
 
 class PropertyEncryptor:
     """
-    Class to mimic the standard Jasypt string encryption/decryption.
+    Provides property value encryption/decryption support via PBEWithMD5AndDES, which is the
+    default encryption algorithm used by Jasypt's CLI and StandardPBEByteEncryptor. This aligns with
+    the same approach used for property encryption within the Krausening Java package.
 
-    Modified from: https://github.com/binsgit/PBEWithMD5AndDES/blob/master/python/PBEWithMD5AndDES_2.py
+    As per https://fermenter.atlassian.net/browse/KRAUS-11, later iterations should seek to introduce
+    a more secure encryption algorithm approach.
 
     See https://bitbucket.org/cpointe/krausening/src/dev/ for details on encrypting values with Jasypt.
     """
 
-    def encrypt(self, msg: str, password: bytes) -> bytes:
+    def encrypt(self, value_to_encrypt: str, password: bytes) -> bytes:
         salt = os.urandom(8)
-        pad_num = 8 - (len(msg) % 8)
-        for i in range(pad_num):
-            msg += chr(pad_num)
-        (dk, iv) = self.get_derived_key(password, salt, 1000)
-        crypter = DES.new(dk, DES.MODE_CBC, iv)
-        enc_text = crypter.encrypt(msg)
-        return base64.b64encode(salt + enc_text)
-
-    def decrypt(self, msg: str, password: bytes) -> str:
-        msg_bytes = base64.b64decode(msg)
+        (key, init_vector) = self._pbkdf1_md5(password, salt, 1000)
+        cipher = Cipher(TripleDES(key), CBC(init_vector))
+        encryptor = cipher.encryptor()
+        return encryptor.update(value_to_encrypt) + encryptor.finalize()
+
+    def decrypt(self, encrypted_msg: str, password: bytes) -> str:
+        msg_bytes = base64.b64decode(encrypted_msg)
         salt = msg_bytes[:8]
         enc_text = msg_bytes[8:]
-        (dk, iv) = self.get_derived_key(password, salt, 1000)
-        crypter = DES.new(dk, DES.MODE_CBC, iv)
-        text = crypter.decrypt(enc_text)
+        (key, init_vector) = self._pbkdf1_md5(password, salt, 1000)
+
+        cipher = Cipher(TripleDES(key), CBC(init_vector))
+        decryptor = cipher.decryptor()
+        text = decryptor.update(enc_text) + decryptor.finalize()
         # remove the padding at the end, if any
         return re.sub(r"[\x01-\x08]", "", text.decode("utf-8"))
 
-    def get_derived_key(self, password: bytes, salt: bytes, count: int) -> tuple:
-        key = password + salt
-        for i in range(count):
-            m = hashlib.md5(key)
-            key = m.digest()
-        return (key[:8], key[8:])
+    def _pbkdf1_md5(self, password, salt, iterations):
+        """
+        Provides a Password Based Key Derivation Function (PBKDF1) as defined in RFC 2829
+        (https://www.rfc-editor.org/rfc/rfc2898#section-5.1) that applies a MD5 hash function
+        to derive a key from the given password.
+        """
+        digest = Hash(MD5())
+        digest.update(password)
+        digest.update(salt)
+
+        key = None
+        for i in range(iterations):
+            key = digest.finalize()
+            digest = Hash(MD5())
+            digest.update(key)
+
+        digest.finalize()
+
+        return key[:8], key[8:16]

krausening-python/src/krausening/properties/property_encryptor.py

@@ -83,7 +83,8 @@ def getProperty(self, key: str, defaultValue: Optional[T] = None):
 class EncryptableProperties(Properties):
     """
     This class represents a properties file that can decrypt property values that have been encrypted
-    with Jasypt, and is based on Jaspyt's EncryptableProperties.
+    via PBEWithMD5AndDES, which is the default encryption algorithm used by Jasypt's CLI and StandardPBEByteEncryptor.
+    This class is largely based on Jaspyt's EncryptableProperties.
 
     See https://bitbucket.org/cpointe/krausening/src/dev/ for details on encrypting values with Jasypt.
     """

krausening-python/src/krausening/properties/property_manager.py

@@ -1,7 +1,10 @@
 import os
 
 
-def after_feature(context, feature):
-    if "properties" == feature.name:
-        os.environ["KRAUSENING_BASE"] = None
-        os.environ["KRAUSENING_EXTENSION"] = None
+def before_scenario(context, scenario):
+    """
+    Clear all Krausening environment variables prior to each scenario.
+    """
+    os.environ["KRAUSENING_BASE"] = ""
+    os.environ["KRAUSENING_EXTENSIONS"] = ""
+    os.environ["KRAUSENING_PASSWORD"] = ""

krausening-python/tests/features/environment.py

@@ -2,16 +2,18 @@
 Feature: Property Management
 
   Scenario: Property can be loaded from file
-    Given a base file with property "foo" and value "bar"
-    When the property file is loaded
-    Then the value of "foo" is set to "bar"
+    Given a base properties file with property "foo"
+    When the properties file is loaded
+    Then the retrieved value of "foo" is "bar"
 
   Scenario: Properties can be overridden
-    Given a base file with property "foo" and value "bar"
-    When a new property file is read with property "foo" and value "bar2"
-    Then the property value is set to "bar2"
+    Given a base properties file with property "foo"
+    And an extensions properties file with property "foo"
+    When the properties file is loaded
+    Then the retrieved value of "foo" is "bar2"
 
   Scenario: Encrypted properties can be decrypted
-    Given a base file with property "foo" and an encrypted value for "bar"
-    When the property file is loaded
-    Then the value of "foo" is set to "bar"
+    Given a base properties file with property "foo"
+    And the properties file contains encrypted value for the "foo" property
+    When the properties file is loaded
+    Then the retrieved value of "foo" is "bar"

krausening-python/tests/features/properties.feature

@@ -2,44 +2,46 @@
 
 from behave import *
 from krausening.properties import PropertyManager
+from nose.tools import assert_equal
 
-use_step_matcher("re")
 
-propertyManager = PropertyManager.get_instance()
-properties = None
-
-
-@given('a base file with property "foo" and value "bar"')
+@given('a base properties file with property "foo"')
 def step_impl(context):
     os.environ["KRAUSENING_BASE"] = "tests/resources/config/"
     context.file = "test.properties"
 
 
-@given('a base file with property "foo" and an encrypted value for "bar"')
+@given('an extensions properties file with property "foo"')
 def step_impl(context):
-    os.environ["KRAUSENING_BASE"] = "tests/resources/config/"
-    os.environ["KRAUSENING_PASSWORD"] = "P455w0rd"
-    context.file = "test-encrypted.properties"
+    os.environ["KRAUSENING_EXTENSIONS"] = "tests/resources/config_extension/"
 
 
-@when("the property file is loaded")
+@given('the properties file contains encrypted value for the "foo" property')
 def step_impl(context):
-    global properties
-    properties = propertyManager.get_properties(context.file)
+    os.environ["KRAUSENING_PASSWORD"] = "P455w0rd"
+    context.file = "test-encrypted.properties"
 
 
-@then('the value of "foo" is set to "bar"')
+@when("the properties file is loaded")
 def step_impl(context):
-    assert properties["foo"] == "bar"
+    context.properties = PropertyManager.get_instance().get_properties(context.file)
 
 
-@when('a new property file is read with property "foo" and value "bar2"')
+@then('the retrieved value of "foo" is "bar"')
 def step_impl(context):
-    os.environ["KRAUSENING_EXTENSIONS"] = "tests/resources/config_extension/"
-    global properties
-    properties = propertyManager.get_properties("test.properties")
+    foo_property_value = context.properties["foo"]
+    assert_equal(
+        foo_property_value,
+        "bar",
+        f"Retrieved 'foo' property, which is {foo_property_value}, didn't match expected value",
+    )
 
 
-@then('the property value is set to "bar2"')
+@then('the retrieved value of "foo" is "bar2"')
 def step_impl(context):
-    assert properties["foo"] == "bar2"
+    foo_property_value = context.properties["foo"]
+    assert_equal(
+        foo_property_value,
+        "bar2",
+        f"Retrieved 'foo' property, which is {foo_property_value}, didn't match expected value",
+    )