1. rsaUtil.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# File : rsaUtil.py
# Author: DaShenHan& long ----- First bitter, then sweet , Let the evening wind blow the willow face ------
# Date : 2021-04-16
import json
from Crypto.Cipher import PKCS1_v1_5 as Cipher_pkcs1_v1_5
from Crypto.PublicKey import RSA
import base64
class documentRsa():
__public_key = "MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCmuYcy3TlrCbh4n34VKFt0swBBXYy5sHMkKr91nYJjoG98kbPgR2n6iYWoU9mI/Zi0geVX2b3YV1Y9Aq/5l5DoJVyG52ZNmnKxtD8ZYJFnah3/Lr8YEGsvXxjv41mxgEQYuoa+OTcd1KOUmeP8R+ZheJj6KhFmhaU40ocjSVXDTQIDAQAB"
__private_key = "MIICXQIBAAKBgQCmuYcy3TlrCbh4n34VKFt0swBBXYy5sHMkKr91nYJjoG98kbPgR2n6iYWoU9mI/Zi0geVX2b3YV1Y9Aq/5l5DoJVyG52ZNmnKxtD8ZYJFnah3/Lr8YEGsvXxjv41mxgEQYuoa+OTcd1KOUmeP8R+ZheJj6KhFmhaU40ocjSVXDTQIDAQABAoGAU8FiLOg7vwzsEhS1hzV19EM0PqIbXvFIdoPPGOveo+Pl/eNIRMQ7djP95BjzcEFNpGiY6TAmkfcWCKI/+xfsIBNoz2JU/JRQbtLGGLOHYgH2mDWLoKVDGAS8Cree1ogjYMP3SrnhqN61/OvFNBlCR72JieQJf6sgYaQUWtRtw1UCQQDWD3+5dBkaisQiCZC9I1jOM86qd0rRpZPa/OPjHRZuaFi1bsBDniPO2wEgLrHo4iZQDDvIx8X2JJR0KLjLBq/LAkEAx2PV+T9HWHukbF+C2y5gpeDrZbMo3A2rsN26j4f5f48XCwaxhc+SaINOGDUb2mwJdRl9FzKREwFNCTOOkwzGRwJAOzhweosfAtPPLSd+ToU/5VQ54bk9DsEDJbt/BG8grIzbtd4c/Nf8ShcWuaORfr9KIC0yQPrbzpwSIgy+CxUffwJBAKAt8xLmJAHQannpjO1Z56i69UDhWhzWGqzeo+Z0vRodsMxZ1zmBVMQDYDSQTcz95+cbz+NX6zYSYHnY4lv98jcCQQC7IhqbN+qLZwMPz26JN0F01HgbKYJ2LLmM6QXrUGRKUolEK5VF2QwC7t8dbPNSY7M8RpE4GQja6kUvY0qAbIGP"
def __init__(self, rsa_publicKey, rsa_privateKey):
"""
Class initialization , Pass in config Two fields configured in the object , The public and private keys stored internally are text without a start
:param rsa_publicKey: config.rsa_publicKey
:param rsa_privateKey: config.rsa_privateKey
"""
self._public_key = rsa_publicKey
self._private_key = rsa_privateKey
def get_public_key(self):
"""
Publicly obtain the complete public key
:return:
"""
return "-----BEGIN PUBLIC KEY-----\n" + self._public_key + "\n-----END PUBLIC KEY-----"
def get_private_key(self):
"""
Publicly obtain and splice the complete private key
:return:
"""
return "-----BEGIN RSA PRIVATE KEY-----\n" + self._private_key + "\n-----END RSA PRIVATE KEY-----"
@staticmethod
def _rsa_decrypt(priv_key_str, msg):
"""
Static methods , Decrypt the string according to the private key
:param priv_key_str: Private key integrity
:param msg: Public key encryption string
:return:
"""
msg = base64.b64decode(msg)
length = len(msg)
default_length = 128
# Private key decryption
priobj = Cipher_pkcs1_v1_5.new(RSA.importKey(priv_key_str))
# The length does not need to be segmented
if length < default_length:
return b''.join(priobj.decrypt(msg, b' '))
# It needs to be segmented
offset = 0
res = []
while length - offset > 0:
if length - offset > default_length:
res.append(priobj.decrypt(msg[offset:offset + default_length], b' '))
else:
res.append(priobj.decrypt(msg[offset:], b' '))
offset += default_length
return b''.join(res)
@staticmethod
def _rsa_encrypt(pub_key_str, msg):
"""
Static methods , Encrypt the string according to the public key
:param pub_key_str: The public key is complete
:param msg: To be encrypted json Text
:return:
"""
msg = msg.encode('utf-8')
length = len(msg)
default_length = 117
# default_length = 100
# Public key encryption
pubobj = Cipher_pkcs1_v1_5.new(RSA.importKey(pub_key_str))
# The length does not need to be segmented
if length < default_length:
return base64.b64encode(pubobj.encrypt(msg))
# It needs to be segmented
offset = 0
res = []
while length - offset > 0:
if length - offset > default_length:
res.append(pubobj.encrypt(msg[offset:offset + default_length]))
else:
res.append(pubobj.encrypt(msg[offset:]))
offset += default_length
byte_data = b''.join(res)
return base64.b64encode(byte_data)
def rsa_decrypt(self, msg):
"""
Decrypt text to b’ Binary value
:param msg:
:return:
"""
return self._rsa_decrypt(self.get_private_key(), msg)
def rsa_decrypt2str(self, msg):
"""
Get decrypted text
:param msg: Encrypted text
:return:
"""
return self.rsa_decrypt(msg).decode("utf-8") # Decrypt into byte And turn to utf-8 character string
def rsa_decrypt2dict(self, msg):
"""
Get decryption json Text to dict
:param msg: Encrypted text
:return:
"""
return json.loads(self.rsa_decrypt2str(msg)) # Get the transmitted data dict
def rsa_encrypt(self, msg, pub_key=None):
"""
Encrypts a string into b' Binary system
:param msg: Text to be encrypted
:param pub_key: Public key , Do not fill in the default get initialized
:return:
"""
pub_key = pub_key or self.get_public_key()
return self._rsa_encrypt(pub_key, msg)
def rsa_encrypt2str(self, msg):
"""
Encrypt text to text
:param msg: Text to be encrypted
:return:
"""
return self.rsa_encrypt(msg).decode("utf-8") # Encrypted into a string to be transmitted
2.rsa_test.py, Pay attention to yourself import Of .utils.rsaUtil Location for the core class file above
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# File : rsa_test.py
# Author: DaShenHan& long ----- First bitter, then sweet , Let the evening wind blow the willow face ------
# Date : 2021-04-16
import sys, os
from urllib.parse import quote, unquote
# path = os.path.dirname(__file__) + os.sep + '../'
# sys.path.append(path)
from winbao.document_pro.utils.rsaUtil import documentRsa
import json
from time import time, strftime
from datetime import datetime, timedelta
# test rsa encryption
def test():
__public_key = "MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCmuYcy3TlrCbh4n34VKFt0swBBXYy5sHMkKr91nYJjoG98kbPgR2n6iYWoU9mI/Zi0geVX2b3YV1Y9Aq/5l5DoJVyG52ZNmnKxtD8ZYJFnah3/Lr8YEGsvXxjv41mxgEQYuoa+OTcd1KOUmeP8R+ZheJj6KhFmhaU40ocjSVXDTQIDAQAB"
__private_key = "MIICXQIBAAKBgQCmuYcy3TlrCbh4n34VKFt0swBBXYy5sHMkKr91nYJjoG98kbPgR2n6iYWoU9mI/Zi0geVX2b3YV1Y9Aq/5l5DoJVyG52ZNmnKxtD8ZYJFnah3/Lr8YEGsvXxjv41mxgEQYuoa+OTcd1KOUmeP8R+ZheJj6KhFmhaU40ocjSVXDTQIDAQABAoGAU8FiLOg7vwzsEhS1hzV19EM0PqIbXvFIdoPPGOveo+Pl/eNIRMQ7djP95BjzcEFNpGiY6TAmkfcWCKI/+xfsIBNoz2JU/JRQbtLGGLOHYgH2mDWLoKVDGAS8Cree1ogjYMP3SrnhqN61/OvFNBlCR72JieQJf6sgYaQUWtRtw1UCQQDWD3+5dBkaisQiCZC9I1jOM86qd0rRpZPa/OPjHRZuaFi1bsBDniPO2wEgLrHo4iZQDDvIx8X2JJR0KLjLBq/LAkEAx2PV+T9HWHukbF+C2y5gpeDrZbMo3A2rsN26j4f5f48XCwaxhc+SaINOGDUb2mwJdRl9FzKREwFNCTOOkwzGRwJAOzhweosfAtPPLSd+ToU/5VQ54bk9DsEDJbt/BG8grIzbtd4c/Nf8ShcWuaORfr9KIC0yQPrbzpwSIgy+CxUffwJBAKAt8xLmJAHQannpjO1Z56i69UDhWhzWGqzeo+Z0vRodsMxZ1zmBVMQDYDSQTcz95+cbz+NX6zYSYHnY4lv98jcCQQC7IhqbN+qLZwMPz26JN0F01HgbKYJ2LLmM6QXrUGRKUolEK5VF2QwC7t8dbPNSY7M8RpE4GQja6kUvY0qAbIGP"
rsa_obj = documentRsa(__public_key, __private_key)
expiry = 2
expiry_time_str = strftime("%Y%m%d %H:%M:%S", (datetime.now() + timedelta(hours=expiry + 8)).timetuple())
print(expiry_time_str)
print(quote(expiry_time_str))
msg_dict = {
'bucket': 'winbao',
'object_name': ' Account function optimization requirements 2020.12.18.docx',
'expiry': expiry_time_str # 20210416 13:39:22
}
mtime = datetime.strptime(expiry_time_str, '%Y%m%d %H:%M:%S')
print(' The time-out time passed :', mtime)
print(' Timeout :', datetime.now() + timedelta(hours=8) > mtime)
print(' Calculated timeout :', datetime.now() + timedelta(hours=expiry + 8))
t1 = time()
msg = json.dumps(msg_dict, ensure_ascii=False)
data_encode_str = rsa_obj.rsa_encrypt2str(msg)
t2 = time()
print(' Encrypted string :', data_encode_str)
print(' Encrypted string parameters url code :', quote(data_encode_str))
print(' Encryption takes time (s):', t2 - t1)
print(len(data_encode_str))
data_decode_str = rsa_obj.rsa_decrypt2str(data_encode_str)
print(data_decode_str)
if __name__ == '__main__':
test()
3. Running results :
