Merge pull request #104 from cmason3/dev

update crypto stuff

Author: cmason3

CHANGELOG.md

@@ -1,5 +1,9 @@
 ## CHANGELOG
 
+### [1.23.1] - December 12, 2024
+- Renamed class `Vault` to `AnsibleVault`
+- Moved class `Vaulty` from `ext_jinjafx.py` into `jinjafx.py`
+
 ### [1.23.0] - December 10, 2024
 - Added support for multiple templates within a DataTemplate to allow templates to be nested
 
@@ -548,6 +552,7 @@ Would result in the following:
 - Initial release
 
 
+[1.23.1]: https://github.com/cmason3/jinjafx/compare/v1.23.0...v1.23.1
 [1.23.0]: https://github.com/cmason3/jinjafx/compare/v1.22.2...v1.23.0
 [1.22.2]: https://github.com/cmason3/jinjafx/compare/v1.22.1...v1.22.2
 [1.22.1]: https://github.com/cmason3/jinjafx/compare/v1.22.0...v1.22.1

extensions/ext_jinjafx.py

@@ -18,9 +18,7 @@
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
 from cryptography.hazmat.primitives.kdf.scrypt import Scrypt
-from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305
-from cryptography.exceptions import InvalidTag
-from jinjafx import JinjaFx
+from jinjafx import JinjaFx, Vaulty
 
 import os, base64, random, re, hashlib, ipaddress
 
@@ -314,51 +312,3 @@ def __xpath(self, s_xml, s_path):
 
     else:
       raise JinjaFx.TemplateError("'xpath' filter requires the 'lxml' python module")
-
-class Vaulty():
-  def __init__(self):
-    self.__prefix = '$VAULTY;'
-    self.__kcache = {}
-
-  def __derive_key(self, password, salt=None):
-    if (ckey := (password, salt)) in self.__kcache:
-      e = self.__kcache[ckey]
-      self.__kcache[ckey] = e[0], e[1], e[2] + 1
-      return self.__kcache[ckey]
-
-    if salt is None:
-      salt = os.urandom(16)
-  
-    key = Scrypt(salt, 32, 2**16, 8, 1).derive(password.encode('utf-8'))
-    self.__kcache[ckey] = [salt, key, 0]
-    return [salt, key, 0]
-
-  def encrypt(self, plaintext, password, cols=None):
-    version = b'\x01'
-    salt, key, uc = self.__derive_key(password)
-    nonce = os.urandom(12)[:-4] + uc.to_bytes(4, 'big')
-    ciphertext = ChaCha20Poly1305(key).encrypt(nonce, plaintext.encode('utf-8'), None)
-
-    r = self.__prefix + base64.b64encode(version + salt + nonce + ciphertext).decode('utf-8')
-   
-    if cols is not None:
-      r = '\n'.join([r[i:i + cols] for i in range(0, len(r), cols)])
-
-    return r
-  
-  def decrypt(self, ciphertext, password):
-    if ciphertext.lstrip().startswith(self.__prefix):
-      try:
-        nciphertext = base64.b64decode(ciphertext.strip()[len(self.__prefix):])
-
-        if len(nciphertext) > 29 and nciphertext.startswith(b'\x01'):
-          key = self.__derive_key(password, nciphertext[1:17])[1]
-          return ChaCha20Poly1305(key).decrypt(nciphertext[17:29], nciphertext[29:], None).decode('utf-8')
-
-      except Exception:
-        pass
-
-      raise JinjaFx.TemplateError('invalid vaulty password or ciphertext malformed')
-
-    raise JinjaFx.TemplateError('data not encrypted with vaulty')
-

jinjafx.py

@@ -20,20 +20,24 @@
   sys.exit('Requires Python >= 3.9')
 
 import os, io, importlib.util, argparse, re, getpass, datetime, traceback, copy
-import jinja2, jinja2.sandbox, yaml, zoneinfo
+import jinja2, jinja2.sandbox, yaml, zoneinfo, base64
 
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
+from cryptography.hazmat.primitives.kdf.scrypt import Scrypt
 from cryptography.hazmat.primitives.hmac import HMAC
 from cryptography.hazmat.primitives.padding import PKCS7
 from cryptography.hazmat.primitives.ciphers import Cipher
 from cryptography.hazmat.primitives.ciphers.algorithms import AES
 from cryptography.hazmat.primitives.ciphers.modes import CTR
+from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305
+
 from cryptography.exceptions import InvalidSignature
+from cryptography.exceptions import InvalidTag
 
-__version__ = '1.23.0'
+__version__ = '1.23.1'
 
-__all__ = ['JinjaFx', 'Vault']
+__all__ = ['JinjaFx', 'AnsibleVault', 'Vaulty']
 
 def main():
   exc_source = None
@@ -105,7 +109,7 @@ def main():
           raise Exception('multiline stings not permitted')
 
         __get_vault_credentials(vpw, True)
-        vtext = Vault().encrypt(b_string, vpw[0])
+        vtext = AnsibleVault().encrypt(b_string, vpw[0])
         print('!vault |\n' + re.sub(r'^', ' ' * 10, vtext, flags=re.MULTILINE))
 
       else:
@@ -117,7 +121,7 @@ def main():
 
             try:
               with open(f, 'rb') as fh:
-                vtext = Vault().encrypt(fh.read(), vpw[0])
+                vtext = AnsibleVault().encrypt(fh.read(), vpw[0])
 
               with open(f, 'wb') as fh:
                 fh.write(vtext.encode('utf-8'))
@@ -137,7 +141,7 @@ def main():
       if not args.decrypt:
         b_vtext = sys.stdin.buffer.read()
         __get_vault_credentials(vpw)
-        print(Vault().decrypt(b_vtext, vpw[0]).decode('utf-8'))
+        print(AnsibleVault().decrypt(b_vtext, vpw[0]).decode('utf-8'))
 
       else:
         __get_vault_credentials(vpw)
@@ -148,7 +152,7 @@ def main():
 
             try:
               with open(f, 'rb') as fh:
-                plaintext = Vault().decrypt(fh.read(), vpw[0])
+                plaintext = AnsibleVault().decrypt(fh.read(), vpw[0])
 
               with open(f, 'wb') as fh:
                 fh.write(plaintext)
@@ -387,7 +391,7 @@ def main():
 def __decrypt_vault(vpw, string):
   if string.lstrip().startswith('$ANSIBLE_VAULT;'):
     __get_vault_credentials(vpw)
-    return Vault().decrypt(string.encode('utf-8'), vpw[0])
+    return AnsibleVault().decrypt(string.encode('utf-8'), vpw[0])
   return string.encode('utf-8')
 
 
@@ -1324,7 +1328,7 @@ def __jfx_now(self, fmt=None, tz='UTC'):
       return str(datetime.datetime.now(tz=zoneinfo.ZoneInfo(tz)))
 
 
-class Vault():
+class AnsibleVault():
   def __derive_key(self, b_password, b_salt=None):
     if b_salt is None:
       b_salt = os.urandom(32)
@@ -1386,5 +1390,53 @@ def decrypt(self, b_string, password):
       raise Exception('data isn\'t ansible vault encrypted')
 
 
+class Vaulty():
+  def __init__(self):
+    self.__prefix = '$VAULTY;'
+    self.__kcache = {}
+
+  def __derive_key(self, password, salt=None):
+    if (ckey := (password, salt)) in self.__kcache:
+      e = self.__kcache[ckey]
+      self.__kcache[ckey] = e[0], e[1], e[2] + 1
+      return self.__kcache[ckey]
+
+    if salt is None:
+      salt = os.urandom(16)
+
+    key = Scrypt(salt, 32, 2**16, 8, 1).derive(password.encode('utf-8'))
+    self.__kcache[ckey] = [salt, key, 0]
+    return [salt, key, 0]
+
+  def encrypt(self, plaintext, password, cols=None):
+    version = b'\x01'
+    salt, key, uc = self.__derive_key(password)
+    nonce = os.urandom(12)[:-4] + uc.to_bytes(4, 'big')
+    ciphertext = ChaCha20Poly1305(key).encrypt(nonce, plaintext.encode('utf-8'), None)
+
+    r = self.__prefix + base64.b64encode(version + salt + nonce + ciphertext).decode('utf-8')
+
+    if cols is not None:
+      r = '\n'.join([r[i:i + cols] for i in range(0, len(r), cols)])
+
+    return r
+
+  def decrypt(self, ciphertext, password):
+    if ciphertext.lstrip().startswith(self.__prefix):
+      try:
+        nciphertext = base64.b64decode(ciphertext.strip()[len(self.__prefix):])
+
+        if len(nciphertext) > 29 and nciphertext.startswith(b'\x01'):
+          key = self.__derive_key(password, nciphertext[1:17])[1]
+          return ChaCha20Poly1305(key).decrypt(nciphertext[17:29], nciphertext[29:], None).decode('utf-8')
+
+      except Exception:
+        pass
+
+      raise JinjaFx.TemplateError('invalid vaulty password or ciphertext malformed')
+
+    raise JinjaFx.TemplateError('data not encrypted with vaulty')
+
+
 if __name__ == '__main__':
   main()