diff --git a/appPublic/rsa.py b/appPublic/rsa.py deleted file mode 100644 index fb5aa38..0000000 --- a/appPublic/rsa.py +++ /dev/null @@ -1,176 +0,0 @@ -from cryptography.hazmat.backends import default_backend -from cryptography.hazmat.primitives.asymmetric import rsa -from cryptography.hazmat.primitives import hashes -from cryptography.hazmat.primitives.asymmetric import padding -from cryptography.hazmat.primitives import serialization -from cryptography.exceptions import InvalidSignature -import base64 - -class RSA: - def __init__(self): - pass - - def write_privatekey(self,private_key,fname,password=None): - pwd = password - pem = '' - if pwd is not None: - pwd = bytes(pwd,encoding='utf8') if not isinstance(pwd, bytes) else pwd - pem = private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.BestAvailableEncryption(pwd) - ) - else: - pem = private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.TraditionalOpenSSL, - encryption_algorithm=serialization.NoEncryption() - ) - - - with open(fname,'w') as f: - text = pem.decode('utf8') - f.write(text) - - def publickeyText(self,public_key): - pem = public_key.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo - ) - text = pem.decode('utf8') - return text - - def write_publickey(self,public_key,fname): - text = self.publickeyText(public_key) - with open(fname,'w') as f: - f.write(text) - - def read_privatekey(self,fname,password=None): - pwd = password - if password is not None: - pwd = bytes(password,encoding='utf8') if not isinstance(password, bytes) else password - with open(fname, "rb") as key_file: - key = serialization.load_pem_private_key( - key_file.read(), - password=pwd, - backend=default_backend() - ) - return key - - def publickeyFromText(self,text): - public_key_bytes = bytes(text,encoding='utf8') - return serialization.load_pem_public_key(data=public_key_bytes,backend=default_backend()) - - def read_publickey(self,fname): - with open(fname,'r') as f: - text = f.read() - return self.publickeyFromText(text) - - def create_privatekey(self, keylength=2048): - return rsa.generate_private_key( - public_exponent=65537, - key_size=keylength, - backend=default_backend() - ) - - def create_publickey(self,private_key): - return private_key.public_key() - - def encode_bytes(self, public_key, bdata): - return public_key.encrypt(bdata, - padding.OAEP( - mgf=padding.MGF1(algorithm=hashes.SHA256()), - algorithm=hashes.SHA256(), - label=None - )) - - def encode(self,public_key,text): - message_bytes = bytes(text, encoding='utf8') if not isinstance(text, bytes) else text - cdata = self.encode_bytes(public_key, message_bytes) - return str(base64.b64encode(cdata), encoding='utf-8') - - def decode_bytes(self, private_key, bdata): - return private_key.decrypt( - bdata,padding.OAEP( - mgf=padding.MGF1(algorithm=hashes.SHA256()), - algorithm=hashes.SHA256(), - label=None - ) - ) - - def decode(self,private_key,cipher): - cipher = cipher.encode('utf8') if not isinstance(cipher, bytes) else cipher - ciphertext_decoded = base64.b64decode(cipher) - plain_text = self.decode_bytes(private_key, ciphertext_decoded) - return str(plain_text, encoding='utf8') - - def sign_bdata(self, private_key, data_to_sign): - s = private_key.sign(data_to_sign, - padding.PSS( - mgf=padding.MGF1(hashes.SHA256()), - salt_length=padding.PSS.MAX_LENGTH - ), - hashes.SHA256() - ) - return s - - def sign(self,private_key,message): - data_to_sign = bytes(message, encoding='utf8') if not isinstance( - message, - bytes - ) else message - signature = str( - base64.b64encode(self.sign_bdata(private_key, data_to_sign)), - encoding='utf8' - ) - return signature - - def check_sign_bdata(self, public_key, bdata, sign): - try: - r = public_key.verify(sign, bdata, - padding.PSS( - mgf=padding.MGF1(hashes.SHA256()), - salt_length=padding.PSS.MAX_LENGTH - ), - hashes.SHA256() - ) - # print('verify return=', r) r is None - return True - except InvalidSignature as e: - return False - - def check_sign(self,public_key,plain_text,signature): - plain_text_bytes = bytes( - plain_text, - encoding='utf8' - ) if not isinstance(plain_text, bytes) else plain_text - signature = base64.b64decode( - signature - ) if not isinstance(signature, bytes) else signature - return self.check_sign_bdata(public_key, plain_text_bytes, \ - signature) - -if __name__ == '__main__': - import os - prikey1_file = os.path.join(os.path.dirname(__file__),'..','test', 'prikey1.rsa') - r = RSA() - mpri = r.create_privatekey(4096) - mpub = r.create_publickey(mpri) - - zpri = r.create_privatekey(4096) - zpub = r.create_publickey(zpri) - - l = 100 - while True: - text = 'h' * l - cipher = r.encode(mpub,text) - ntext = r.decode(mpri,cipher) - print('textlen=', l, 'encode text=', text, \ - 'decode result=', ntext, - 'cyber size=', len(cipher), - 'check if equal=', text==ntext) - signature = r.sign(zpri,text) - check = r.check_sign(zpub,text,signature) - print('sign and verify=',len(signature),check) - l += 1 - diff --git a/appPublic/rsawrap.py b/appPublic/rsawrap.py new file mode 100644 index 0000000..9e030a1 --- /dev/null +++ b/appPublic/rsawrap.py @@ -0,0 +1,105 @@ +import rsa + +class RSA: + + def __init__(self, keylength=4096, coding='iso8859'): + self.coding = coding + self.keylength = keylength + + def write_privatekey(self,private_key,fname,password=None): + bd = private_key.save_pkcs1() + with open(fname, 'wb') as f: + f.write(bd) + + def publickeyText(self,public_key): + bd = public_key.save_pkcs1() + return bd.encode(self.coding) + + def write_publickey(self,public_key,fname): + bd = public_key.save_pkcs1() + with open(fname, 'wb') as f: + f.write(bd) + + def read_privatekey(self,fname,password=None): + with open(fname, 'rb') as pf: + kd = pf.read() + return rsa.PrivateKey.load_pkcs1(kd) + + def publickeyFromText(self,text): + bd = text.encode(self.coding) + rsa.PublicKey.load_pkcs1(bd) + + def read_publickey(self,fname): + with open(fname, 'rb') as pf: + kd = pf.read() + return rsa.PublicKey.load_pkcs1(kd) + + def create_privatekey(self, keylength=4096): + _, prik = rsa.newkeys(keylength) + return prik + + def create_publickey(self,private_key): + return rsa.PublicKey(private_key.n, private_key.e) + + def encode_bytes(self, public_key, bdata): + return rsa.encrypt(bdata, public_key) + + def encode(self,public_key,text): + bdata = text.encode(self.coding) + bc = self.encode_bytes(public_key, bdata) + return bc.decode(self.coding) + + def decode_bytes(self, private_key, bdata): + return rsa.decrypt(bdata, private_key) + + def decode(self,private_key,cipher): + bc = cipher.encode(self.coding) + bd = self.decode_bytes(private_key, bc) + return bd.decode(self.coding) + + def sign_bdata(self, private_key, data_to_sign): + return rsa.sign(data_to_sign, private_key, 'SHA-1') + + def sign(self,private_key,message): + bd = message.encode(self.coding) + bs = self.sign_bdata(private_key, bd) + return bs.decode(self.coding) + + def check_sign_bdata(self, public_key, bdata, sign): + try: + r = rsa.verify(bdata, sign, public_key) + if r == 'SHA-1': + return True + except: + return False + + + def check_sign(self,public_key,plain_text,signature): + bd = plain_text.encode(self.coding) + bs = signature.encode(self.coding) + return self.check_sign_bdata(public_key, bd, bs) + + +if __name__ == '__main__': + import os + prikey1_file = os.path.join(os.path.dirname(__file__),'..','test', 'prikey1.rsa') + r = RSA() + mpri = r.create_privatekey(2048) + mpub = r.create_publickey(mpri) + + zpri = r.create_privatekey(2048) + zpub = r.create_publickey(zpri) + + l = 100 + while True: + text = 'h' * l + cipher = r.encode(mpub,text) + ntext = r.decode(mpri,cipher) + print('textlen=', l, 'encode text=', text, \ + 'decode result=', ntext, + 'cyber size=', len(cipher), + 'check if equal=', text==ntext) + signature = r.sign(zpri,text) + check = r.check_sign(zpub,text,signature) + print('sign and verify=',len(signature),check) + l += 1