JWT Authentication and RabbitMQ for Microservices

Posted on Dec 18, 2024 in Computers

JWT Authentication for a Secure REST API

Server-Side Implementation

import jwt
from flask import Flask, request, jsonify


app = Flask(__name__)

SECRET_KEY = 'coen4246313'

# Secured API route
@app.route('/secure-data', methods=['GET'])
def secure_data():
    token = request.headers.get('Authorization')

    if not token:
        return jsonify({"error": "Missing token"}), 401

    try:
        # Extract the token from the 'Bearer' scheme
        token = token.split(" ")[1]

        # Validate the token
        decoded_payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
        return jsonify({"message": "Access granted", "data": decoded_payload})

    except jwt.ExpiredSignatureError:
        return jsonify({"error": "Token has expired"}), 401
    except jwt.InvalidTokenError:
        return jsonify({"error": "Invalid token"}), 401

if __name__ == '__main__':
    app.run(debug=True)

Client-Side Implementation

import requests
import datetime
import jwt
# Define the API endpoint
url = 'http://127.0.0.1:5000/secure-data'

# Your JWT token (replace with the actual token generated earlier)
#token = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxMjMsInJvbGUiOiJhZG1pbiIsImV4cCI6MTcyODM1MTY1Mn0.Nh3ZyAoL7BReIGsGMcuRsH3wSZF295b__BKiMNdNCXc'

# Secret key used for signing the JWT
SECRET_KEY = 'coen4246313'

# Define the payload (data)
payload = {
    'user_id': 123,
    'role': 'admin',
    'exp': datetime.datetime.utcnow() + datetime.timedelta(hours=1)  # Token expiration time
}

# Define the headers (algorithm)
headers = {
    'alg': 'HS256',
    'typ': 'JWT'
}

# Generate the JWT
token = jwt.encode(payload, SECRET_KEY, algorithm='HS256', headers=headers)

print(f"Generated JWT: {token}")

# Token validation
try:
    # Decode and validate the token
    decoded_payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])

    print(f"Token is valid. Decoded payload: {decoded_payload}")

except jwt.ExpiredSignatureError:
    print("Token has expired.")
except jwt.InvalidTokenError:
    print("Invalid token.")


# Define headers with the Authorization token
headers = {
    'Authorization': f'Bearer {token}'
}

# Send the GET request
response = requests.get(url, headers=headers)

# Print the response
if response.status_code == 200:
    print("Access granted:", response.json())
else:
    print(f"Access denied: {response.status_code}, {response.json()}")

Microservice Communication with RabbitMQ

Order Service (Consumer)

import pika
import json

# RabbitMQ connection and channel setup
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Declare the same queue to listen for events
channel.queue_declare(queue='user_updates')

# Function to simulate updating the user’s email in the Order Service’s database
def update_email_in_order_service(user_id, new_email):
    print(f"Updating email for user {user_id} in Order Service to {new_email}")

# Callback function to process messages from RabbitMQ
def callback(ch, method, properties, body):
    event_data = json.loads(body)
    print(f"[x] Received user update event: {event_data}")
    # Update email in the Order Service’s database
    update_email_in_order_service(event_data['user_id'], event_data['new_email'])

# Setup the consumer to listen to the 'user_updates' queue
channel.basic_consume(queue='user_updates', on_message_callback=callback, auto_ack=True)

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


channel.start_consuming()

User Service (Producer)

import pika
import json

# RabbitMQ connection and channel setup
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Declare a queue named 'user_updates' for publishing events
channel.queue_declare(queue='user_updates')

# Function to simulate email update in the User Service and publish the event
def update_user_email(user_id, new_email):
    # Simulate updating the user’s email in the User Service’s database
    print(f"User {user_id} email updated to {new_email} in User Service.")

    # Create an event to notify other services about this update
    event_data = {
        "user_id": user_id,
        "new_email": new_email
    }

    # Publish the event to RabbitMQ
    message = json.dumps(event_data)
    channel.basic_publish(exchange='', routing_key='user_updates', body=message)
    print(f" [x] Published user update event: {event_data}")

# Example: Simulate a user email update
update_user_email("user_123", "newemail@example.com")

# Close the connection
connection.close()