python使用RSA加密算法

全栈程序员站长

发布于 2022-08-20 02:14:15

2.7K00

代码可运行

文章被收录于专栏：全栈程序员必看全栈程序员必看

运行总次数：0

代码可运行

大家好，又见面了，我是你们的朋友全栈君。

上一篇文章介绍了RSA加密原理以及自己的一些理解，现在我们就来实际操作一下，使用python语言如何来实现RSA的加密—解密—签名—验签这一系列过程。

##一、生成公钥、私钥对使用openssl工具生成RSA公钥、私钥对。

1.下载openssl工具。点我下载一，点我下载二 2.打开 openssl 文件夹下的 bin 文件夹，执行 openssl.exe 文件； 3.生成RSA私钥命令：

这里生成的密钥文件是2048比特。

genrsa -out rsa_private_key.pem 2048

4.生成RSA公钥命令： 注意，公钥和私钥是成对的，所以你在生成一个后，另一个的生成是基于前一个的文件名，否则生成的不是一对哦！！

rsa -in rsa_private_key.pem -pubout -out rsa_public_key.pem

生成后的文件应该在你的用户目录下，或者在openssl的bin文件夹下（有时候在有时又不在，真是搞不懂，反正你就在这两个地方都找一下，肯定在的）。

**5.这里介绍一下RSA密钥文件的规律，**文件名都是以.pem为后缀，生成的RSA密钥，可以看到都是以-----BEGIN RSA PRIVATE KEY-----开头，-----END RSA PRIVATE KEY-----结尾的没有换行的字符串，这个就是原始的 RSA 私钥。

##二、安装python支持的加密库—pycryptodome

我用的python3.6版本，网上搜了一下，在python3.6之前的版本大部分是用pycrypto来进行加密的，而在python3.6之后，因无人维护pycrypto，所以改名为pycryptodome，这个库的强大之处在于，他是pycrypto库的扩展，用起来更加方便，更灵活啊~~功能杠杠滴

###1.安装pycryptodome 如果你安装了pip，那么就很简单了。

pip3 install pycryptodome

###2.将密钥文件导入到程序中。

from Crypto.PublicKey import RSA

# 读取标准的rsa公私钥pem文件
def load_rsa_file(fn):
    key = None
    try:
        key = RSA.importKey(open(fn).read())
    except Exception as err:
        print('导入rsa的KEY文件出错', fn, err)
    return key


# 标准字符串密钥转rsa格式密钥
def rsa_key_str2std(skey):
    ret = None
    try:
        ret = RSA.importKey(skey)
    except Exception as err:
        print('字符串密钥转rsa格式密钥错误', skey, err)
    return ret

###3.RSA加密

from Crypto.Cipher import PKCS1_OAEP

# RSA_加密
def rsa_enc(data, rsa_key):
    ciphertext = b''
    try:
        cipher = PKCS1_OAEP.new(rsa_key)
        ciphertext = cipher.encrypt(data)
    except Exception as err:
        print('RSA加密失败', '', err)
    return ciphertext

由于RSA在加密过程中，每次加密只能加密最大长度的字符串，如果你的加密数据超长，在加密过程中需要分段加密，同理，解密也是分段解密的。

1024位的证书，加密时最大支持117个字节，解密时为128； 2048位的证书，加密时最大支持245个字节，解密时为256。

加密时支持的最大字节数：证书位数/8 -11（比如：2048位的证书，支持的最大加密字节数：2048/8 – 11 = 245）其中，11位字节为保留字节。上面我的密钥文件是2048比特，所以加密分块长度为245字节。

###4.加密分块

# 根据key长度计算分块大小
def get_block_size(rsa_key):
	try:
		# RSA仅支持限定长度内的数据的加解密，需要分块
		# 分块大小block_reversed_size=11
		reserve_size = block_reversed_size
		key_size = rsa_key.size_in_bits()
		if (key_size % 8) != 0:
			raise RuntimeError('RSA 密钥长度非法')

		# 密钥用来解密，解密不需要预留长度
		if rsa_key.has_private():
			reserve_size = 0

		bs = int(key_size / 8) - reserve_size
	except Exception as err:
		print('计算加解密数据块大小出错', rsa_key, err)
	return bs

# 返回块数据
def block_data(data, rsa_key):
	bs = get_block_size(rsa_key)
	for i in range(0, len(data), bs):
		yield data[i:i + bs]

RSA在解密分段时与加密时用的分段大小无关，都是按照密钥长度/8来分段解密的。

###5.RSA解密

from Crypto.Cipher import PKCS1_OAEP

# RSA解密
def rsa_dec(data, rsa_key):
    ret_data = b''
    try:
        cipher = PKCS1_OAEP.new(rsa_key)
        ret_data = cipher.decrypt(data)
    except Exception as err:
        print('RSA解密失败', '', err)
    return ret_data

###6.RSA签名

from Crypto.Signature import pkcs1_15
from Crypto.Hash import SHA256

# RSA签名
def rsa_sign(data, rsa_key):
    signature = ''
    try:
        h = SHA256.new(data)
        signature = pkcs1_15.new(rsa_key).sign(h)
    except Exception as err:
        print('RSA签名失败', '', err)
    return signature

###7.RSA验签名

# RSA签名验证
def rsa_sign_verify(data, sig, rsa_key):
    try:
        h = SHA256.new(data)
        pkcs1_15.new(rsa_key).verify(h, sig)
        ret = True
    except (ValueError, TypeError):
        ret = False
    return ret

###8.RSA加解密类（我将RSA加解密封装成一个类，方便后续直接调用），完整代码如下：

# -*- coding: utf-8 -*-
import Crypto.Cipher as Cipher
import Crypto.Signature as Sign
import Crypto.Hash as Hash
from Crypto.PublicKey import RSA
from Crypto.Cipher import PKCS1_v1_5 as PKCS1_v1_5_cipper
from Crypto.Signature import PKCS1_v1_5 as PKCS1_v1_5_sign
from Crypto.Hash import SHA1


class Rsa:
    """RSA加解密签名类
    """

    def __int__(self, ciper_lib=PKCS1_v1_5_cipper, sign_lib=PKCS1_v1_5_sign, hash_lib=SHA1,
                pub_file=None, pri_file=None, pub_skey=None, pri_skey=None, pub_key=None, pri_key=None,
                reversed_size=11):

        # 加解密库
        self.ciper_lib = ciper_lib
        self.sign_lib = sign_lib
        self.hash_lib = hash_lib

        # 公钥密钥
        if pub_key:
            self.pub_key = pub_key
        elif pub_skey:
            self.pub_key = RSA.importKey(pub_skey)
        elif pub_file:
            self.pub_key = RSA.importKey(open(pub_file).read())

        if pri_key:
            self.pri_key = pri_key
        elif pri_skey:
            self.pri_key = RSA.importKey(pri_skey)
        elif pri_file:
            self.pri_key = RSA.importKey(open(pri_file).read())

        # 分块保留长度
        self.block_reversed_size = reversed_size

    # 根据key长度计算分块大小
    def get_block_size(self, rsa_key):
        try:
            # RSA仅支持限定长度内的数据的加解密，需要分块
            # 分块大小
            reserve_size = self.block_reversed_size
            key_size = rsa_key.size_in_bits()
            if (key_size % 8) != 0:
                raise RuntimeError('RSA 密钥长度非法')

            # 密钥用来解密，解密不需要预留长度
            if rsa_key.has_private():
                reserve_size = 0

            bs = int(key_size / 8) - reserve_size
        except Exception as err:
            print('计算加解密数据块大小出错', rsa_key, err)
        return bs

    # 返回块数据
    def block_data(self, data, rsa_key):
        bs = self.get_block_size(rsa_key)
        for i in range(0, len(data), bs):
            yield data[i:i + bs]

    # 加密
    def enc_bytes(self, data, key=None):
        text = b''
        try:
            rsa_key = self.pub_key
            if key:
                rsa_key = key

            cipher = self.ciper_lib.new(rsa_key)
            for dat in self.block_data(data, rsa_key):
                cur_text = cipher.encrypt(dat)
                text += cur_text
        except Exception as err:
            print('RSA加密失败', data, err)
        return text

    # 解密
    def dec_bytes(self, data, key=None):
        text = b''
        try:
            rsa_key = self.pri_key
            if key:
                rsa_key = key

            cipher = self.ciper_lib.new(rsa_key)
            for dat in self.block_data(data, rsa_key):
                if type(self.ciper_lib) == Cipher.PKCS1_OAEP:
                    cur_text = cipher.decrypt(dat)
                else:
                    cur_text = cipher.decrypt(dat, '解密异常')
                text += cur_text
        except Exception as err:
            print('RSA解密失败', data, err)
        return text

    # RSA签名
    def sign_bytes(self, data, key=None):
        signature = ''
        try:
            rsa_key = self.pri_key
            if key:
                rsa_key = key

            h = self.hash_lib.new(data)
            signature = self.sign_lib.new(rsa_key).sign(h)
        except Exception as err:
            print('RSA签名失败', '', err)
        return signature

    # RSA签名验证
    def sign_verify(self, data, sig, key=None):
        try:
            rsa_key = self.pub_key
            if key:
                rsa_key = key
            h = self.hash_lib.new(data)
            self.sign_lib.new(rsa_key).verify(h, sig)
            ret = True
        except (ValueError, TypeError):
            ret = False
        return ret


def main():
    pass


if __name__ == '__main__':
    main()