统一消息系统

import pika

import json

from Crypto.Cipher import AES

# 消息加密函数

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):

cipher = AES.new(key, AES.MODE_EAX, nonce=encrypted_data['nonce'])

decrypted = cipher.decrypt_and_verify(encrypted_data['ciphertext'], encrypted_data['tag'])

return decrypted.decode('utf-8')

# 发送消息

def send_message():

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

channel = connection.channel()

channel.queue_declare(queue='secure_queue')

message = {"user": "admin", "action": "login", "timestamp": "2025-04-10T10:30:00Z"}

key = b'your-secure-key-1234567890123456'

encrypted = encrypt_message(json.dumps(message), key)

channel.basic_publish(

exchange='',

routing_key='secure_queue',

body=json.dumps(encrypted)

)

print(" [x] Sent encrypted message")

connection.close()

# 接收消息

def receive_message():

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

channel = connection.channel()

channel.queue_declare(queue='secure_queue')

def callback(ch, method, properties, body):

encrypted_data = json.loads(body)

key = b'your-secure-key-1234567890123456'

decrypted = decrypt_message(encrypted_data, key)

print(" [x] Received:", decrypted)

channel.basic_consume(callback, queue='secure_queue', no_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')

channel.start_consuming()

# 运行发送和接收

if __name__ == "__main__":

send_message()

# receive_message() # 可以单独运行接收端