oci/zeronet
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Revert "Switch to sslcrypto for cryptography tasks (#2338)"

Browse source 
This reverts commit fbc7b6fc4f.
This commit is contained in:
shortcutme 2019-12-15 18:30:42 +01:00
parent 2f7323043f
commit 958882c1c5
No known key found for this signature in database
GPG key ID: 5B63BAE6CB9613AE
55 changed files with 7287 additions and 3748 deletions
Download patch file Download diff file Expand all files Collapse all files

56
plugins/CryptMessage/CryptMessagePlugin.py
View file

@ -5,22 +5,24 @@ import gevent
from Plugin import PluginManager
from Crypt import CryptBitcoin, CryptHash
import lib.pybitcointools as btctools
from Config import config
import sslcrypto
from . import CryptMessage
curve = sslcrypto.ecc.get_curve("secp256k1")
@PluginManager.registerTo("UiWebsocket")
class UiWebsocketPlugin(object):
def eciesDecrypt(self, encrypted, privatekey):
back = CryptMessage.getEcc(privatekey).decrypt(encrypted)
return back.decode("utf8")
# - 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))
publickey = self.user.getEncryptPublickey(self.site.address, index)
self.response(to, publickey)
# Encrypt a text using the publickey or user's sites unique publickey
# Return: Encrypted text using base64 encoding
@ -53,16 +55,23 @@ class UiWebsocketPlugin(object):
# Encrypt a text using AES
# Return: Iv, AES key, Encrypted text
def actionAesEncrypt(self, to, text, key=None):
def actionAesEncrypt(self, to, text, key=None, iv=None):
from lib import pyelliptic
if key:
key = base64.b64decode(key)
else:
key = sslcrypto.aes.new_key()
key = os.urandom(32)
if iv: # Generate new AES key if not definied
iv = base64.b64decode(iv)
else:
iv = pyelliptic.Cipher.gen_IV('aes-256-cbc')
if text:
encrypted, iv = sslcrypto.aes.encrypt(text.encode("utf8"), key)
encrypted = pyelliptic.Cipher(key, iv, 1, ciphername='aes-256-cbc').ciphering(text.encode("utf8"))
else:
encrypted, iv = b"", b""
encrypted = b""
res = [base64.b64encode(item).decode("utf8") for item in [key, iv, encrypted]]
self.response(to, res)
@ -70,6 +79,8 @@ class UiWebsocketPlugin(object):
# Decrypt a text using AES
# Return: Decrypted text
def actionAesDecrypt(self, to, *args):
from lib import pyelliptic
if len(args) == 3: # Single decrypt
encrypted_texts = [(args[0], args[1])]
keys = [args[2]]
@ -82,8 +93,9 @@ class UiWebsocketPlugin(object):
iv = base64.b64decode(iv)
text = None
for key in keys:
ctx = pyelliptic.Cipher(base64.b64decode(key), iv, 0, ciphername='aes-256-cbc')
try:
decrypted = sslcrypto.aes.decrypt(encrypted_text, iv, base64.b64decode(key))
decrypted = ctx.ciphering(encrypted_text)
if decrypted and decrypted.decode("utf8"): # Valid text decoded
text = decrypted.decode("utf8")
except Exception as err:
@ -110,11 +122,12 @@ class UiWebsocketPlugin(object):
# Gets the publickey of a given privatekey
def actionEccPrivToPub(self, to, privatekey):
self.response(to, curve.private_to_public(curve.wif_to_private(privatekey)))
self.response(to, btctools.privtopub(privatekey))
# Gets the address of a given publickey
def actionEccPubToAddr(self, to, publickey):
self.response(to, curve.public_to_address(bytes.fromhex(publickey)))
address = btctools.pubtoaddr(btctools.decode_pubkey(publickey))
self.response(to, address)
@PluginManager.registerTo("User")
@ -150,7 +163,7 @@ class UserPlugin(object):
if "encrypt_publickey_%s" % index not in site_data:
privatekey = self.getEncryptPrivatekey(address, param_index)
publickey = curve.private_to_public(curve.wif_to_private(privatekey))
publickey = btctools.encode_pubkey(btctools.privtopub(privatekey), "bin_compressed")
site_data["encrypt_publickey_%s" % index] = base64.b64encode(publickey).decode("utf8")
return site_data["encrypt_publickey_%s" % index]
@ -187,8 +200,8 @@ class ActionsPlugin:
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"))
ecc = CryptMessage.getEcc(self.privatekey)
assert ecc.decrypt(encrypted) == self.utf8_text.encode("utf8"), "%s != %s" % (ecc.decrypt(encrypted), self.utf8_text.encode("utf8"))
yield "."
def testCryptEciesDecryptMulti(self, num_run=1):
@ -210,16 +223,23 @@ class ActionsPlugin:
gevent.joinall(threads)
def testCryptAesEncrypt(self, num_run=1):
from lib import pyelliptic
for i in range(num_run):
key = os.urandom(32)
encrypted = sslcrypto.aes.encrypt(self.utf8_text.encode("utf8"), key)
iv = pyelliptic.Cipher.gen_IV('aes-256-cbc')
encrypted = pyelliptic.Cipher(key, iv, 1, ciphername='aes-256-cbc').ciphering(self.utf8_text.encode("utf8"))
yield "."
def testCryptAesDecrypt(self, num_run=1):
from lib import pyelliptic
key = os.urandom(32)
encrypted_text, iv = sslcrypto.aes.encrypt(self.utf8_text.encode("utf8"), key)
iv = pyelliptic.Cipher.gen_IV('aes-256-cbc')
encrypted_text = pyelliptic.Cipher(key, iv, 1, ciphername='aes-256-cbc').ciphering(self.utf8_text.encode("utf8"))
for i in range(num_run):
decrypted = sslcrypto.aes.decrypt(encrypted_text, iv, key).decode("utf8")
ctx = pyelliptic.Cipher(key, iv, 0, ciphername='aes-256-cbc')
decrypted = ctx.ciphering(encrypted_text).decode("utf8")
assert decrypted == self.utf8_text
yield "."