Celery Security: Securing Workers, Brokers, and Task Communications

from celery import Celery
from kombu.utils.url import quote

app = Celery('security_demo')

# Secure Redis connection with TLS and password
app.conf.broker_url = 'rediss://:password@redis.example.com:6380/0'
app.conf.broker_use_ssl = {
    'ssl_cert_reqs': 'required',
    'ssl_ca_certs': '/etc/ssl/certs/ca.crt',
    'ssl_certfile': '/etc/ssl/certs/client.crt',
    'ssl_keyfile': '/etc/ssl/private/client.key',
}

@app.task
def secure_task(data):
    print(f"Executing secure task with: {data}")
    return {"status": "ok", "data": data}

class BrokerSecurityChecker:
    def __init__(self, app):
        self.app = app

    def check_broker_url(self):
        """Analyze broker URL for security issues."""
        url = self.app.conf.broker_url or ""
        issues = []

        if url.startswith("redis://") and ":" not in url.split("@")[0]:
            issues.append("No password set on Redis broker URL")
        elif url.startswith("amqp://") and url.count(":") < 3:
            issues.append("No credentials on RabbitMQ broker URL")

        if url.startswith("redis://"):
            issues.append("Redis connection without TLS (use rediss://)")
        elif url.startswith("amqp://") and "amqps" not in url:
            issues.append("AMQP connection without TLS (use amqps://)")

        if not url:
            issues.append("No broker URL configured")

        return {"broker_url": url, "issues": issues, "secure": len(issues) == 0}

    def check_serializer(self):
        """Check serializer configuration."""
        serializer = self.app.conf.task_serializer
        accepted = self.app.conf.accept_content
        issues = []

        if serializer == "pickle":
            issues.append("CRITICAL: Using pickle serializer - arbitrary code execution risk")
        if "pickle" in accepted:
            issues.append("CRITICAL: Accepting pickle content - arbitrary code execution risk")

        return {"serializer": serializer, "accepted": accepted, "issues": issues}

checker = BrokerSecurityChecker(app)

broker_check = checker.check_broker_url()
print("Broker Security Check:")
for issue in broker_check["issues"]:
    print(f"  ISSUE: {issue}")
print(f"  Overall: {'SECURE' if broker_check['secure'] else 'ISSUES FOUND'}")

serializer_check = checker.check_serializer()
print(f"\nSerializer Check:")
print(f"  Serializer: {serializer_check['serializer']}")
for issue in serializer_check["issues"]:
    print(f"  ISSUE: {issue}")

from celery import Celery
import re

app = Celery('security_demo', broker='redis://localhost:6379/0')

class TaskValidator:
    def __init__(self):
        self.rules = {}

    def add_rule(self, task_name, schema):
        """Add validation rules for a task."""
        self.rules[task_name] = schema

    def validate(self, task_name, args, kwargs):
        """Validate task arguments against schema."""
        if task_name not in self.rules:
            return {"valid": True}

        schema = self.rules[task_name]
        errors = []

        combined_args = {}
        for i, (arg_name, arg_schema) in enumerate(schema.get("positional", {}).items()):
            if i < len(args):
                combined_args[arg_name] = args[i]

        combined_args.update(kwargs)

        for field, rules in schema.get("fields", {}).items():
            value = combined_args.get(field)

            if rules.get("required") and value is None:
                errors.append(f"Field '{field}' is required")
                continue

            if value is not None:
                if "max_length" in rules and len(str(value)) > rules["max_length"]:
                    errors.append(f"Field '{field}' exceeds max length {rules['max_length']}")

                if "pattern" in rules and not re.match(rules["pattern"], str(value)):
                    errors.append(f"Field '{field}' does not match pattern")

                if "type" in rules and not isinstance(value, rules["type"]):
                    errors.append(f"Field '{field}' must be of type {rules['type'].__name__}")

        return {"valid": len(errors) == 0, "errors": errors}

validator = TaskValidator()

validator.add_rule("security_demo.send_notification", {
    "positional": {},
    "fields": {
        "email": {
            "required": True,
            "type": str,
            "max_length": 254,
            "pattern": r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$',
        },
        "message": {
            "required": True,
            "type": str,
            "max_length": 5000,
        },
    }
})

@app.task(bind=True)
def send_notification(self, email, message):
    """Send a notification after validating arguments."""
    result = validator.validate(self.name, [email, message], {})
    if not result["valid"]:
        raise ValueError(f"Validation failed: {', '.join(result['errors'])}")
    print(f"Sending notification to {email}: {message[:50]}...")
    return {"sent": True, "to": email}

send_notification.delay("user@example.com", "Hello from Celery")
import time
time.sleep(0.3)

try:
    send_notification.delay("invalid-email", "Test")
except Exception as e:
    print(f"Validation prevented: {e}")

time.sleep(0.3)