python实现rsa加密

模拟效果：用户注册/登陆成功，返回自己的公钥，注册时候生成的私钥和公钥入库。下次进来的时候，加解密是动态的，也就是每个用户都不一样

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import serialization, hashes
import base64
import sqlite3

# 生成私钥和公钥
def generate_keys():
    private_key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=2048,
        backend=default_backend()
    )
    public_key = private_key.public_key()

    # 将公钥和私钥序列化为PEM格式，并进行Base64编码
    public_key_pem = public_key.public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    )
    private_key_pem = private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption()
    )
    public_key_b64 = base64.b64encode(public_key_pem).decode('utf-8')
    private_key_b64 = base64.b64encode(private_key_pem).decode('utf-8')

    return private_key_b64, public_key_b64

# 将公钥和私钥存储到数据库
def store_keys_in_database(user_id, private_key_b64, public_key_b64):
    conn = sqlite3.connect('example.db')
    c = conn.cursor()
    # 创建用户表，如果已存在则忽略
    c.execute('''CREATE TABLE IF NOT EXISTS users
                 (id INTEGER PRIMARY KEY, private_key TEXT, public_key TEXT)''')
    # 插入公钥和私钥
    c.execute('INSERT INTO users (id, private_key, public_key) VALUES (?, ?, ?)', (user_id, private_key_b64, public_key_b64))
    # 如果是自增id。
    # 获取最后插入行的自增ID
    # last_row_id = c.lastrowid
    # print(f'用户ID: {last_row_id}')
    conn.commit()
    conn.close()

# RSA加密（使用公钥）
def encrypt_with_public_key(public_key_b64, plaintext):
    public_key_pem = base64.b64decode(public_key_b64)
    public_key = serialization.load_pem_public_key(
        public_key_pem,
        backend=default_backend()
    )
    encrypted = public_key.encrypt(
        plaintext,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )
    return base64.b64encode(encrypted).decode('utf-8')

# RSA解密（使用私钥）
def decrypt_with_private_key(private_key_b64, ciphertext_b64):
    ciphertext = base64.b64decode(ciphertext_b64)
    private_key_pem = base64.b64decode(private_key_b64)
    private_key = serialization.load_pem_private_key(
        private_key_pem,
        password=None,
        backend=default_backend()
    )
    decrypted = private_key.decrypt(
        ciphertext,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )
    return decrypted.decode('utf-8')

# 主程序
def main():
    # 生成密钥对
    user_id = 2  # 假设用户ID为1
    private_key_b64, public_key_b64 = generate_keys()
    print("公钥(Base64):", public_key_b64)
    print("私钥(Base64):", private_key_b64)

    # 存储公钥和私钥到数据库，id唯一，主键
    #store_keys_in_database(user_id, private_key_b64, public_key_b64)
    print("生成完成 用户id:", user_id)

    # 从数据库检索公钥和私钥
    conn = sqlite3.connect('example.db')
    c = conn.cursor()
    c.execute('SELECT private_key, public_key FROM users WHERE id = ?', (user_id,))
    stored_private_key_b64, stored_public_key_b64 = c.fetchone()
    conn.close()

    # 要加密的明文
    plaintext = "Hello, World!"

    # 使用公钥加密
    encrypted_text_b64 = encrypt_with_public_key(stored_public_key_b64, plaintext.encode('utf-8'))
    print("加密(Base64):", encrypted_text_b64)

    # 使用私钥解密
    decrypted_text = decrypt_with_private_key(stored_private_key_b64, encrypted_text_b64)
    print("解密:", decrypted_text)

if __name__ == "__main__":
    main()

输出

公钥(Base64): xxxx
私钥(Base64): yyy
生成完成 用户id: 2
加密(Base64): xxyy
解密: Hello, World!