统一消息系统

import jwt

from Crypto.Cipher import AES

from Crypto.Random import get_random_bytes

# 消息加密函数

def encrypt_message(message, key):

cipher = AES.new(key, AES.MODE_EAX)

ciphertext, tag = cipher.encrypt_and_digest(message.encode('utf-8'))

return {

'nonce': cipher.nonce,

'tag': tag,

'ciphertext': ciphertext

}

# 消息解密函数

def decrypt_message(encrypted_data, key):

nonce = encrypted_data['nonce']

tag = encrypted_data['tag']

ciphertext = encrypted_data['ciphertext']

cipher = AES.new(key, AES.MODE_EAX, nonce=nonce)

plaintext = cipher.decrypt_and_verify(ciphertext, tag)

return plaintext.decode('utf-8')

# JWT生成函数

def generate_token(user_id, secret_key):

payload = {

'user_id': user_id,

'type': 'message'

}

token = jwt.encode(payload, secret_key, algorithm='HS256')

return token

# 验证JWT

def verify_token(token, secret_key):

try:

payload = jwt.decode(token, secret_key, algorithms=['HS256'])

return payload.get('user_id')

except jwt.ExpiredSignatureError:

return None

except jwt.InvalidTokenError:

return None

# 示例：发送一条加密消息

message = "这是一条测试消息"

key = get_random_bytes(16) # 128位密钥

encrypted = encrypt_message(message, key)

# 生成JWT令牌

secret_key = 'your-secret-key'

token = generate_token('user123', secret_key)

print("加密后的消息:", encrypted)

print("JWT令牌:", token)

# 假设接收方使用相同密钥和令牌验证

user_id = verify_token(token, secret_key)

if user_id:

decrypted = decrypt_message(encrypted, key)

print("解密后的消息:", decrypted)

else:

print("无效的令牌")