123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226 |
- import base64
- import os
- import gevent
- from Plugin import PluginManager
- from Crypt import CryptBitcoin, CryptHash
- from Config import config
- import sslcrypto
- from . import CryptMessage
- curve = sslcrypto.ecc.get_curve("secp256k1")
- @PluginManager.registerTo("UiWebsocket")
- class UiWebsocketPlugin(object):
- # - Actions -
- # Returns user's public key unique to site
- # Return: Public key
- def actionUserPublickey(self, to, index=0):
- self.response(to, self.user.getEncryptPublickey(self.site.address, index))
- # Encrypt a text using the publickey or user's sites unique publickey
- # Return: Encrypted text using base64 encoding
- def actionEciesEncrypt(self, to, text, publickey=0, return_aes_key=False):
- if type(publickey) is int: # Encrypt using user's publickey
- publickey = self.user.getEncryptPublickey(self.site.address, publickey)
- aes_key, encrypted = CryptMessage.eciesEncrypt(text.encode("utf8"), publickey)
- if return_aes_key:
- self.response(to, [base64.b64encode(encrypted).decode("utf8"), base64.b64encode(aes_key).decode("utf8")])
- else:
- self.response(to, base64.b64encode(encrypted).decode("utf8"))
- # Decrypt a text using privatekey or the user's site unique private key
- # Return: Decrypted text or list of decrypted texts
- def actionEciesDecrypt(self, to, param, privatekey=0):
- if type(privatekey) is int: # Decrypt using user's privatekey
- privatekey = self.user.getEncryptPrivatekey(self.site.address, privatekey)
- if type(param) == list:
- encrypted_texts = param
- else:
- encrypted_texts = [param]
- texts = CryptMessage.eciesDecryptMulti(encrypted_texts, privatekey)
- if type(param) == list:
- self.response(to, texts)
- else:
- self.response(to, texts[0])
- # Encrypt a text using AES
- # Return: Iv, AES key, Encrypted text
- def actionAesEncrypt(self, to, text, key=None):
- if key:
- key = base64.b64decode(key)
- else:
- key = sslcrypto.aes.new_key()
- if text:
- encrypted, iv = sslcrypto.aes.encrypt(text.encode("utf8"), key)
- else:
- encrypted, iv = b"", b""
- res = [base64.b64encode(item).decode("utf8") for item in [key, iv, encrypted]]
- self.response(to, res)
- # Decrypt a text using AES
- # Return: Decrypted text
- def actionAesDecrypt(self, to, *args):
- if len(args) == 3: # Single decrypt
- encrypted_texts = [(args[0], args[1])]
- keys = [args[2]]
- else: # Batch decrypt
- encrypted_texts, keys = args
- texts = [] # Decoded texts
- for iv, encrypted_text in encrypted_texts:
- encrypted_text = base64.b64decode(encrypted_text)
- iv = base64.b64decode(iv)
- text = None
- for key in keys:
- try:
- decrypted = sslcrypto.aes.decrypt(encrypted_text, iv, base64.b64decode(key))
- if decrypted and decrypted.decode("utf8"): # Valid text decoded
- text = decrypted.decode("utf8")
- except Exception as err:
- pass
- texts.append(text)
- if len(args) == 3:
- self.response(to, texts[0])
- else:
- self.response(to, texts)
- # Sign data using ECDSA
- # Return: Signature
- def actionEcdsaSign(self, to, data, privatekey=None):
- if privatekey is None: # Sign using user's privatekey
- privatekey = self.user.getAuthPrivatekey(self.site.address)
- self.response(to, CryptBitcoin.sign(data, privatekey))
- # Verify data using ECDSA (address is either a address or array of addresses)
- # Return: bool
- def actionEcdsaVerify(self, to, data, address, signature):
- self.response(to, CryptBitcoin.verify(data, address, signature))
- # Gets the publickey of a given privatekey
- def actionEccPrivToPub(self, to, privatekey):
- self.response(to, curve.private_to_public(curve.wif_to_private(privatekey.encode())))
- # Gets the address of a given publickey
- def actionEccPubToAddr(self, to, publickey):
- self.response(to, curve.public_to_address(bytes.fromhex(publickey)))
- @PluginManager.registerTo("User")
- class UserPlugin(object):
- def getEncryptPrivatekey(self, address, param_index=0):
- if param_index < 0 or param_index > 1000:
- raise Exception("Param_index out of range")
- site_data = self.getSiteData(address)
- if site_data.get("cert"): # Different privatekey for different cert provider
- index = param_index + self.getAddressAuthIndex(site_data["cert"])
- else:
- index = param_index
- if "encrypt_privatekey_%s" % index not in site_data:
- address_index = self.getAddressAuthIndex(address)
- crypt_index = address_index + 1000 + index
- site_data["encrypt_privatekey_%s" % index] = CryptBitcoin.hdPrivatekey(self.master_seed, crypt_index)
- self.log.debug("New encrypt privatekey generated for %s:%s" % (address, index))
- return site_data["encrypt_privatekey_%s" % index]
- def getEncryptPublickey(self, address, param_index=0):
- if param_index < 0 or param_index > 1000:
- raise Exception("Param_index out of range")
- site_data = self.getSiteData(address)
- if site_data.get("cert"): # Different privatekey for different cert provider
- index = param_index + self.getAddressAuthIndex(site_data["cert"])
- else:
- index = param_index
- if "encrypt_publickey_%s" % index not in site_data:
- privatekey = self.getEncryptPrivatekey(address, param_index).encode()
- publickey = curve.private_to_public(curve.wif_to_private(privatekey) + b"\x01")
- site_data["encrypt_publickey_%s" % index] = base64.b64encode(publickey).decode("utf8")
- return site_data["encrypt_publickey_%s" % index]
- @PluginManager.registerTo("Actions")
- class ActionsPlugin:
- publickey = "A3HatibU4S6eZfIQhVs2u7GLN5G9wXa9WwlkyYIfwYaj"
- privatekey = "5JBiKFYBm94EUdbxtnuLi6cvNcPzcKymCUHBDf2B6aq19vvG3rL"
- utf8_text = '\xc1rv\xedzt\xfbr\xf5t\xfck\xf6rf\xfar\xf3g\xe9p'
- def getBenchmarkTests(self, online=False):
- if hasattr(super(), "getBenchmarkTests"):
- tests = super().getBenchmarkTests(online)
- else:
- tests = []
- aes_key, encrypted = CryptMessage.eciesEncrypt(self.utf8_text.encode("utf8"), self.publickey) # Warm-up
- tests.extend([
- {"func": self.testCryptEciesEncrypt, "kwargs": {}, "num": 100, "time_standard": 1.2},
- {"func": self.testCryptEciesDecrypt, "kwargs": {}, "num": 500, "time_standard": 1.3},
- {"func": self.testCryptEciesDecryptMulti, "kwargs": {}, "num": 5, "time_standard": 0.68},
- {"func": self.testCryptAesEncrypt, "kwargs": {}, "num": 10000, "time_standard": 0.27},
- {"func": self.testCryptAesDecrypt, "kwargs": {}, "num": 10000, "time_standard": 0.25}
- ])
- return tests
- def testCryptEciesEncrypt(self, num_run=1):
- for i in range(num_run):
- aes_key, encrypted = CryptMessage.eciesEncrypt(self.utf8_text.encode("utf8"), self.publickey)
- assert len(aes_key) == 32
- yield "."
- def testCryptEciesDecrypt(self, num_run=1):
- aes_key, encrypted = CryptMessage.eciesEncrypt(self.utf8_text.encode("utf8"), self.publickey)
- for i in range(num_run):
- assert len(aes_key) == 32
- decrypted = CryptMessage.eciesDecrypt(base64.b64encode(encrypted), self.privatekey)
- assert decrypted == self.utf8_text.encode("utf8"), "%s != %s" % (decrypted, self.utf8_text.encode("utf8"))
- yield "."
- def testCryptEciesDecryptMulti(self, num_run=1):
- yield "x 100 (%s threads) " % config.threads_crypt
- aes_key, encrypted = CryptMessage.eciesEncrypt(self.utf8_text.encode("utf8"), self.publickey)
- threads = []
- for i in range(num_run):
- assert len(aes_key) == 32
- threads.append(gevent.spawn(
- CryptMessage.eciesDecryptMulti, [base64.b64encode(encrypted)] * 100, self.privatekey
- ))
- for thread in threads:
- res = thread.get()
- assert res[0] == self.utf8_text, "%s != %s" % (res[0], self.utf8_text)
- assert res[0] == res[-1], "%s != %s" % (res[0], res[-1])
- yield "."
- gevent.joinall(threads)
- def testCryptAesEncrypt(self, num_run=1):
- for i in range(num_run):
- key = os.urandom(32)
- encrypted = sslcrypto.aes.encrypt(self.utf8_text.encode("utf8"), key)
- yield "."
- def testCryptAesDecrypt(self, num_run=1):
- key = os.urandom(32)
- encrypted_text, iv = sslcrypto.aes.encrypt(self.utf8_text.encode("utf8"), key)
- for i in range(num_run):
- decrypted = sslcrypto.aes.decrypt(encrypted_text, iv, key).decode("utf8")
- assert decrypted == self.utf8_text
- yield "."
|