1. rsaUtil.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# File : rsaUtil.py
# Author: DaShenHan&道長-----先苦後甜，任憑晚風拂柳顏------
# 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):
"""
類初始化，傳入config對象內配置的倆字段，內部儲存的公鑰私鑰為不帶開頭的文本
: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):
"""
公開獲取拼接完整的公鑰
:return:
"""
return "-----BEGIN PUBLIC KEY-----\n" + self._public_key + "\n-----END PUBLIC KEY-----"
def get_private_key(self):
"""
公開獲取拼接完整的私鑰
:return:
"""
return "-----BEGIN RSA PRIVATE KEY-----\n" + self._private_key + "\n-----END RSA PRIVATE KEY-----"
@staticmethod
def _rsa_decrypt(priv_key_str, msg):
"""
靜態方法，根據私鑰解密字符串
:param priv_key_str: 私鑰完整
:param msg: 公鑰加密字符串
:return:
"""
msg = base64.b64decode(msg)
length = len(msg)
default_length = 128
# 私鑰解密
priobj = Cipher_pkcs1_v1_5.new(RSA.importKey(priv_key_str))
# 長度不用分段
if length < default_length:
return b''.join(priobj.decrypt(msg, b' '))
# 需要分段
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):
"""
靜態方法,根據公鑰加密字符串
:param pub_key_str: 公鑰完整
:param msg: 待加密json文本
:return:
"""
msg = msg.encode('utf-8')
length = len(msg)
default_length = 117
# default_length = 100
# 公鑰加密
pubobj = Cipher_pkcs1_v1_5.new(RSA.importKey(pub_key_str))
# 長度不用分段
if length < default_length:
return base64.b64encode(pubobj.encrypt(msg))
# 需要分段
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):
"""
解密文本到b’二進制值
:param msg:
:return:
"""
return self._rsa_decrypt(self.get_private_key(), msg)
def rsa_decrypt2str(self, msg):
"""
獲取解密文本
:param msg: 加密後的文本
:return:
"""
return self.rsa_decrypt(msg).decode("utf-8") # 解密成byte並轉為utf-8字符串
def rsa_decrypt2dict(self, msg):
"""
獲取解密json文本到dict
:param msg: 加密後的文本
:return:
"""
return json.loads(self.rsa_decrypt2str(msg)) # 得到傳輸過來的數據的dict
def rsa_encrypt(self, msg, pub_key=None):
"""
加密一段字符串變成 b'二進制
:param msg: 待加密文本
:param pub_key: 公鑰，不填默認獲取初始化的
:return:
"""
pub_key = pub_key or self.get_public_key()
return self._rsa_encrypt(pub_key, msg)
def rsa_encrypt2str(self, msg):
"""
加密文本到文本
:param msg: 待加密文本
:return:
"""
return self.rsa_encrypt(msg).decode("utf-8") # 加密成待傳輸的字符串
2.rsa_test.py,注意自己改import的 .utils.rsaUtil 為上面那個核心類文件放的位置
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# File : rsa_test.py
# Author: DaShenHan&道長-----先苦後甜，任憑晚風拂柳顏------
# 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
# 測試rsa加解密
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': '台賬功能優化需求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('傳過來的超時
