Pad short secrets

README.rst

@@ -98,7 +98,8 @@ for AES256-GCM, with a 16-byte randomly generated initialization vector.
 The encryption produces both the ciphertext as well as a "tag" that is
 used as part of integrity verification. The iv, ciphertext, and tag (in
 that order) are concatenated, base64-encoded, and prepended with ``$7$``
-to produce the encrypted password seen in the configuration files.
+to produce the encrypted password seen in the configuration files. If
+the key is less than 254-bytes it is padded with null bytes.
 
 Phantom
 ~~~~~~~

splunksecrets.py

@@ -62,8 +62,9 @@ def decrypt(secret, ciphertext, nosalt=False):
 
         plaintext = "".join([six.unichr(c) for c in chars])
     elif ciphertext.startswith("$7$"):
-        if len(secret) < 254:
-            raise ValueError(f"secret too short, need 254 bytes, got {len(secret)}")
+        # pad secret to 254 bytes with nulls
+        secret = six.ensure_binary(secret).ljust(254, b"\0")
+
         ciphertext = b64decode(ciphertext[3:])
 
         kdf = PBKDF2HMAC(
@@ -119,8 +120,8 @@ def encrypt(secret, plaintext, nosalt=False):
 
 def encrypt_new(secret, plaintext, iv=None):  # pylint: disable=invalid-name
     """Use the new AES 256 GCM encryption in Splunk 7.2"""
-    if len(secret) < 254:
-        raise ValueError(f"secret too short, need 254 bytes, got {len(secret)}")
+    # pad secret to 254 bytes with nulls
+    secret = six.ensure_binary(secret).ljust(254, b"\0")
 
     kdf = PBKDF2HMAC(
         algorithm=hashes.SHA256(),
@@ -356,6 +357,8 @@ def __load_splunk_secret(ctx, param, value):  # pragma: no cover
     if ctx.get_parameter_source(param.name).name != "ENVIRONMENT":
         with open(value, "rb") as f:  # pylint: disable=invalid-name
             value = f.read().strip()
+    elif isinstance(value, str):
+        value = bytes(value, encoding="utf-8")
 
     return value.strip()
 

tests.py

@@ -105,10 +105,21 @@ def test_encrypt_new(self):
         )
         self.assertEqual(ciphertext, "$7$aTVkS01HYVNJUk5wSnR5NKR+EdOfT4t84WSiXvPFHGHsfHtbgPIL3g==")
 
-    def test_encrypt_new_raises_value_error_short_secret(self):
-        with self.assertRaises(ValueError):
-            splunk_secret = base64.b64encode(os.urandom(255))[:253]
-            splunksecrets.encrypt_new(splunk_secret, "temp1234")
+    def test_encrypt_new_str_secret(self):
+        ciphertext = splunksecrets.encrypt_new(
+            "abc123",  # secret as a string (not bytes)
+            "strings are fine",
+            iv=six.b("zIDM0YmIgDQ2gMzk")
+        )
+        self.assertEqual(ciphertext, "$7$eklETTBZbUlnRFEyZ016a+2HMVEtbCAJkNb7RkVHdqZwZkVJyZ+HmTWlYJedFdR4")
+
+    def test_encrypt_new_pads_short_secret(self):
+        ciphertext = splunksecrets.encrypt_new(
+            splunk_secret[:30],
+            "short123",
+            iv=six.b("4KK0Ra8LWBKxUFQ8")
+        )
+        self.assertEqual(ciphertext, "$7$NEtLMFJhOExXQkt4VUZROK9vm0tDLbJn2jxESMRbs7MTdiHuTtBz8g==")
 
     def test_encrypt_character_matches_salt1(self):
         ciphertext = splunksecrets.encrypt(splunk_secret, "A" * 8)
@@ -131,13 +142,12 @@ def test_decrypt_raises_value_error_short_secret1(self):
             splunk_secret = base64.b64encode(os.urandom(255))[:15]
             splunksecrets.decrypt(splunk_secret, "$1$n6g0W7F51ZAK")
 
-    def test_decrypt_raises_value_error_short_secret2(self):
-        with self.assertRaises(ValueError):
-            splunk_secret = base64.b64encode(os.urandom(255))[:253]
-            splunksecrets.decrypt(
-                splunk_secret,
-                "$7$aTVkS01HYVNJUk5wSnR5NKR+EdOfT4t84WSiXvPFHGHsfHtbgPIL3g=="
-            )
+    def test_decrypt_pads_short_secret2(self):
+        plaintext = splunksecrets.decrypt(
+            splunk_secret[:30],
+            "$7$NEtLMFJhOExXQkt4VUZROK9vm0tDLbJn2jxESMRbs7MTdiHuTtBz8g=="
+        )
+        self.assertEqual(plaintext, "short123")
 
     def test_decrypt_nosalt(self):
         plaintext = splunksecrets.decrypt(splunk_secret, "$1$2+1yGuQ1gcMK", nosalt=True)