Support for encrypted macaroons (#521)

* encrypted macaroons

* fix GRPC env entry

* example config entry

* add pycryptodomex to requirements

* documentation

* Added pycryptodomex to pip file

Co-authored-by: Ben Arc <ben@arc.wales>

lnbits/wallets/lndgrpc.py

@@ -11,6 +11,7 @@ import base64
 import hashlib
 from os import environ, error, getenv
 from typing import Optional, Dict, AsyncGenerator
+from .macaroon import load_macaroon, AESCipher
 
 if imports_ok:
     import lnbits.wallets.lnd_grpc_files.lightning_pb2 as ln
@@ -58,12 +59,6 @@ def get_ssl_context(cert_path: str):
     return context
 
 
-def load_macaroon(macaroon_path: str):
-    with open(macaroon_path, "rb") as f:
-        macaroon_bytes = f.read()
-        return macaroon_bytes.hex()
-
-
 def parse_checking_id(checking_id: str) -> bytes:
     return base64.b64decode(checking_id.replace("_", "/"))
 
@@ -90,18 +85,19 @@ class LndWallet(Wallet):
         self.port = int(getenv("LND_GRPC_PORT"))
         self.cert_path = getenv("LND_GRPC_CERT") or getenv("LND_CERT")
 
-        macaroon_path = (
+        macaroon = (
             getenv("LND_GRPC_MACAROON")
             or getenv("LND_GRPC_ADMIN_MACAROON")
             or getenv("LND_ADMIN_MACAROON")
             or getenv("LND_GRPC_INVOICE_MACAROON")
             or getenv("LND_INVOICE_MACAROON")
         )
-
-        if macaroon_path.split(".")[-1] == "macaroon":
-            self.macaroon = load_macaroon(macaroon_path)
-        else:
-            self.macaroon = macaroon_path
+        
+        
+        encrypted_macaroon = getenv("LND_GRPC_MACAROON_ENCRYPTED")
+        if encrypted_macaroon:
+            macaroon = AESCipher(description="macaroon decryption").decrypt(encrypted_macaroon)    
+        self.macaroon = load_macaroon(macaroon)
 
         cert = open(self.cert_path, "rb").read()
         creds = grpc.ssl_channel_credentials(cert)

lnbits/wallets/lndrest.py

@@ -1,4 +1,5 @@
 import asyncio
+from pydoc import describe
 import httpx
 import json
 import base64
@@ -6,6 +7,7 @@ from os import getenv
 from typing import Optional, Dict, AsyncGenerator
 
 from lnbits import bolt11 as lnbits_bolt11
+from .macaroon import load_macaroon, AESCipher
 
 from .base import (
     StatusResponse,
@@ -34,7 +36,13 @@ class LndRestWallet(Wallet):
             or getenv("LND_INVOICE_MACAROON")
             or getenv("LND_REST_INVOICE_MACAROON")
         )
-        self.auth = {"Grpc-Metadata-macaroon": macaroon}
+
+        encrypted_macaroon = getenv("LND_REST_MACAROON_ENCRYPTED")
+        if encrypted_macaroon:
+            macaroon = AESCipher(description="macaroon decryption").decrypt(encrypted_macaroon)    
+        self.macaroon = load_macaroon(macaroon)
+        
+        self.auth = {"Grpc-Metadata-macaroon": self.macaroon}
         self.cert = getenv("LND_REST_CERT")
 
     async def status(self) -> StatusResponse:

lnbits/wallets/macaroon/__init__.py

@@ -0,0 +1 @@
+from .macaroon import load_macaroon, AESCipher

lnbits/wallets/macaroon/macaroon.py

@@ -0,0 +1,103 @@
+from Cryptodome import Random
+from Cryptodome.Cipher import AES
+import base64
+from hashlib import md5
+import getpass
+
+BLOCK_SIZE = 16
+import getpass 
+
+def load_macaroon(macaroon: str) -> str:
+    """Returns hex version of a macaroon encoded in base64 or the file path. 
+
+    :param macaroon: Macaroon encoded in base64 or file path.
+    :type macaroon: str
+    :return: Hex version of macaroon.
+    :rtype: str
+    """
+
+    # if the macaroon is a file path, load it
+    if macaroon.split(".")[-1] == "macaroon":
+        with open(macaroon, "rb") as f:
+            macaroon_bytes = f.read()
+            return macaroon_bytes.hex()
+    else:
+        # convert the bas64 macaroon to hex
+        try:
+            macaroon = base64.b64decode(macaroon).hex()
+        except:
+            pass
+    return macaroon
+
+class AESCipher(object):
+    """This class is compatible with crypto-js/aes.js
+
+    Encrypt and decrypt in Javascript using:
+        import AES from "crypto-js/aes.js";
+        import Utf8 from "crypto-js/enc-utf8.js";
+        AES.encrypt(decrypted, password).toString()
+        AES.decrypt(encrypted, password).toString(Utf8);
+
+    """
+    def __init__(self, key=None, description=""):
+        self.key = key
+        self.description = description + " "
+
+    def pad(self, data):
+        length = BLOCK_SIZE - (len(data) % BLOCK_SIZE)
+        return data + (chr(length) * length).encode()
+
+
+    def unpad(self, data):
+        return data[: -(data[-1] if type(data[-1]) == int else ord(data[-1]))]
+
+    @property
+    def passphrase(self):
+        passphrase = self.key if self.key is not None else None
+        if passphrase is None:
+            passphrase = getpass.getpass(f"Enter {self.description}password:")
+        return passphrase
+
+    def bytes_to_key(self, data, salt, output=48):
+        # extended from https://gist.github.com/gsakkis/4546068
+        assert len(salt) == 8, len(salt)
+        data += salt
+        key = md5(data).digest()
+        final_key = key
+        while len(final_key) < output:
+            key = md5(key + data).digest()
+            final_key += key
+        return final_key[:output]
+
+    def decrypt(self, encrypted: str) -> str:
+        """Decrypts a string using AES-256-CBC.
+        """
+        passphrase = self.passphrase
+        encrypted = base64.b64decode(encrypted)
+        assert encrypted[0:8] == b"Salted__"
+        salt = encrypted[8:16]
+        key_iv = self.bytes_to_key(passphrase.encode(), salt, 32 + 16)
+        key = key_iv[:32]
+        iv = key_iv[32:]
+        aes = AES.new(key, AES.MODE_CBC, iv)
+        try:
+            return self.unpad(aes.decrypt(encrypted[16:])).decode()
+        except UnicodeDecodeError:
+            raise ValueError("Wrong passphrase")
+
+    def encrypt(self, message: bytes) -> str:
+        passphrase = self.passphrase
+        salt = Random.new().read(8)
+        key_iv = self.bytes_to_key(passphrase.encode(), salt, 32 + 16)
+        key = key_iv[:32]
+        iv = key_iv[32:]
+        aes = AES.new(key, AES.MODE_CBC, iv)
+        return base64.b64encode(b"Salted__" + salt + aes.encrypt(self.pad(message))).decode()
+
+# if this file is executed directly, ask for a macaroon and encrypt it
+if __name__ == "__main__":
+    macaroon = input("Enter macaroon: ")
+    macaroon = load_macaroon(macaroon)
+    macaroon = AESCipher(description="encryption").encrypt(macaroon.encode())
+    print("Encrypted macaroon:")
+    print(macaroon)