apppublic/appPublic/rsa.py

177 lines
5.0 KiB
Python
Raw Normal View History

2019-07-16 16:33:07 +08:00
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)
2019-09-16 15:11:14 +08:00
def publickeyText(self,public_key):
2019-07-16 16:33:07 +08:00
pem = public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
2019-09-16 15:11:14 +08:00
text = pem.decode('utf8')
return text
def write_publickey(self,public_key,fname):
text = self.publickeyText(public_key)
2019-07-16 16:33:07 +08:00
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
2019-09-16 15:20:20 +08:00
def publickeyFromText(self,text):
public_key_bytes = bytes(text,encoding='utf8')
return serialization.load_pem_public_key(data=public_key_bytes,backend=default_backend())
2019-07-16 16:33:07 +08:00
def read_publickey(self,fname):
with open(fname,'r') as f:
2019-09-16 15:20:20 +08:00
text = f.read()
return self.publickeyFromText(text)
2019-07-16 16:33:07 +08:00
2022-03-18 12:45:57 +08:00
def create_privatekey(self, keylength=2048):
2019-07-16 16:33:07 +08:00
return rsa.generate_private_key(
public_exponent=65537,
2022-03-18 12:45:57 +08:00
key_size=keylength,
2019-07-16 16:33:07 +08:00
backend=default_backend()
)
def create_publickey(self,private_key):
return private_key.public_key()
2021-07-13 16:13:59 +08:00
def encode_bytes(self, public_key, bdata):
return public_key.encrypt(bdata,
2019-07-16 16:33:07 +08:00
padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None
2021-07-13 16:13:59 +08:00
))
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')
2019-07-16 16:33:07 +08:00
2021-07-13 16:13:59 +08:00
def decode_bytes(self, private_key, bdata):
return private_key.decrypt(
bdata,padding.OAEP(
2019-07-16 16:33:07 +08:00
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None
)
)
2021-07-13 16:13:59 +08:00
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)
2019-07-16 16:33:07 +08:00
return str(plain_text, encoding='utf8')
2021-07-13 16:13:59 +08:00
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
2019-07-16 16:33:07 +08:00
def sign(self,private_key,message):
data_to_sign = bytes(message, encoding='utf8') if not isinstance(
message,
bytes
) else message
signature = str(
2021-07-13 16:13:59 +08:00
base64.b64encode(self.sign_bdata(private_key, data_to_sign)),
2019-07-16 16:33:07 +08:00
encoding='utf8'
)
return signature
2021-07-13 16:13:59 +08:00
def check_sign_bdata(self, public_key, bdata, sign):
2019-07-16 16:33:07 +08:00
try:
2021-07-13 16:13:59 +08:00
r = public_key.verify(sign, bdata,
2019-07-16 16:33:07 +08:00
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
2021-07-13 16:13:59 +08:00
# print('verify return=', r) r is None
2019-07-16 16:33:07 +08:00
return True
except InvalidSignature as e:
return False
2021-07-13 16:13:59 +08:00
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)
2019-07-16 16:33:07 +08:00
if __name__ == '__main__':
2021-07-13 16:13:59 +08:00
import os
prikey1_file = os.path.join(os.path.dirname(__file__),'..','test', 'prikey1.rsa')
2019-07-16 16:33:07 +08:00
r = RSA()
2022-03-18 12:45:57 +08:00
mpri = r.create_privatekey(4096)
2019-07-16 16:33:07 +08:00
mpub = r.create_publickey(mpri)
2022-03-18 12:45:57 +08:00
zpri = r.create_privatekey(4096)
2019-07-16 16:33:07 +08:00
zpub = r.create_publickey(zpri)
2022-03-18 12:45:57 +08:00
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