Authentication and Authorization in Dramatiq Applications

What Are Authentication and Authorization in Dramatiq?

In a Dramatiq-based system, authentication is the process of verifying the identity of a client or service that interacts with the broker, worker, or web dashboard. Authorization determines whether that verified identity has permission to enqueue a specific task, execute a task, or access administrative endpoints. While Dramatiq itself does not enforce any security policies by default, the surrounding infrastructure (Redis, RabbitMQ, web servers, and middleware) can be configured to protect your background job pipeline.

Why It Matters

Without proper authentication and authorization, your Dramatiq application is vulnerable to:

Unauthorized task injection: Attackers can enqueue malicious tasks that steal data, run arbitrary code, or disrupt operations.
Data leakage: Sensitive arguments passed to tasks (e.g., API keys, personal data) can be intercepted if the broker is unsecured.
Resource exhaustion: Unauthenticated clients can flood the broker with tasks, causing denial of service.
Privilege escalation: Without authorization checks, any authenticated user might run tasks intended only for administrators.

Implementing robust authentication and authorization protects your infrastructure, maintains data integrity, and ensures compliance with security standards.

How to Use Authentication and Authorization in Dramatiq

1. Securing the Broker Connection

The first line of defense is to require credentials for connecting to your message broker (Redis, RabbitMQ, etc.). Below is an example using Redis with a password.

# config.py
import dramatiq
from dramatiq.brokers.redis import RedisBroker

# Use environment variables for secrets
REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_PASSWORD = "your_redis_password"

redis_broker = RedisBroker(
    host=REDIS_HOST,
    port=REDIS_PORT,
    password=REDIS_PASSWORD,
    # Additional SSL/TLS options can be added here
)

dramatiq.set_broker(redis_broker)

For RabbitMQ, you can set a username and password:

from dramatiq.brokers.rabbitmq import RabbitmqBroker

broker = RabbitmqBroker(
    url="amqp://user:password@localhost:5672//"
)

2. Task-Level Authorization Using Middleware

Dramatiq middleware allows you to intercept tasks before execution. You can implement a custom middleware that checks whether the task is allowed to run based on the current user or a token passed in the message.

Example: Authorization Middleware

# middleware/authorization.py
import dramatiq
from dramatiq.middleware import Middleware

class AuthorizationMiddleware(Middleware):
    def before_process_message(self, broker, message):
        """
        This is called before a worker processes a message.
        We can check a custom header (e.g., 'role') and deny execution.
        """
        task_role = message.options.get("required_role", "user")
        # In a real app, you'd retrieve the authenticated user from context
        # For demonstration, we assume a global 'current_user_role' is set.
        from flask import g  # hypothetical
        user_role = getattr(g, "user_role", None)

        if user_role is None or not self._has_permission(user_role, task_role):
            # Reject the task – it will be discarded or moved to DLQ
            raise dramatiq.middleware.Interrupt(
                "Insufficient permissions to execute task"
            )
        return None

    @staticmethod
    def _has_permission(user_role, required_role):
        roles_hierarchy = {"admin": 3, "editor": 2, "user": 1}
        return roles_hierarchy.get(user_role, 0) >= roles_hierarchy.get(required_role, 0)

# Register middleware
broker.add_middleware(AuthorizationMiddleware())

Using the middleware in a task:

# tasks.py
import dramatiq

@dramatiq.actor
def send_report(report_id: int):
    # Task logic
    print(f"Sending report {report_id}")

# When enqueuing, specify required role:
send_report.send_with_options(
    args=(42,),
    options={"required_role": "admin"}
)

3. Decorator-Based Authorization

If you prefer a simpler approach, you can use a decorator that checks permissions inside the task function itself. This is useful when the authorization logic depends on the task arguments (e.g., user ID).

# decorators.py
from functools import wraps
from dramatiq import actor

def require_role(required_role: str):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            # Simulate retrieving the current user from a request context
            # In a real app, you would fetch the user from a token or session
            current_user_role = get_current_user_role()  # your custom function
            if current_user_role != required_role:
                raise PermissionError(f"User role {current_user_role} cannot run this task")
            return fn(*args, **kwargs)
        return actor(wrapper)  # wrap as dramatiq actor
    return decorator

@require_role("admin")
def delete_user(user_id: int):
    # dangerous operation
    print(f"Deleting user {user_id}")

4. Securing the Web Dashboard

Dramatiq provides a web dashboard for monitoring tasks. If you serve it via Flask or FastAPI, you must add authentication. Below is an example using Flask and Flask-Login.

# app.py
from flask import Flask, render_template, redirect, url_for, request
from flask_login import LoginManager, UserMixin, login_user, login_required, current_user
from dramatiq import get_broker
from dramatiq.web import run_dashboard_app

app = Flask(__name__)
app.secret_key = "your-secret-key"
login_manager = LoginManager()
login_manager.init_app(app)

# Simple user model
class User(UserMixin):
    def __init__(self, id, username, password):
        self.id = id
        self.username = username
        self.password = password

users = {"admin": User(1, "admin", "secret")}

@login_manager.user_loader
def load_user(user_id):
    for user in users.values():
        if user.id == int(user_id):
            return user
    return None

@app.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "POST":
        username = request.form["username"]
        password = request.form["password"]
        if username in users and users[username].password == password:
            login_user(users[username])
            return redirect(url_for("dashboard"))
    return render_template("login.html")

@app.route("/dashboard")
@login_required
def dashboard():
    # Run the Dramatiq dashboard app as a Flask blueprint
    dash_app =