Adds CELERY_SECURITY_SERIALIZER and CELERY_SECURITY_DIGEST settings

celery/app/defaults.py

@@ -144,6 +144,8 @@ NAMESPACES = {
         "SECURITY_KEY": Option(None, type="string"),
         "SECURITY_CERTIFICATE": Option(None, type="string"),
         "SECURITY_CERT_STORE": Option(None, type="string"),
+        "SECURITY_DIGEST": Option("sha1", type="string"),
+        "SECURITY_SERIALIZER": Option("pickle", type="string"),
     },
     "CELERYD": {
         "AUTOSCALER": Option("celery.worker.autoscale.Autoscaler"),

celery/security/__init__.py

@@ -1,46 +1,71 @@
 from __future__ import absolute_import
 from __future__ import with_statement
 
-from kombu.serialization import unregister, SerializerNotInstalled
+import warnings
+
+from kombu.serialization import registry, SerializerNotInstalled
 
 from .. import current_app
 from ..exceptions import ImproperlyConfigured
 
 from .serialization import register_auth
 
+SSL_NOT_INSTALLED = """\
+You need to install the pyOpenSSL library to use the auth serializer.
+Please install by:
+
+    $ pip install pyOpenSSL
+"""
+AUTH_DISABLED = """\
+setup_security called, but not configured to use auth serializer.
+Please set CELERY_TASK_SERIALIZER="auth" to enable security.\
+"""
+
+SETTING_MISSING = """\
+Sorry, but you have to configure the
+    * CELERY_SECURITY_KEY
+    * CELERY_SECURITY_CERTIFICATE, and the
+    * CELERY_SECURITY_CERT_STORE
+configuration settings to use the auth serializer.
+
+Please see the configuration reference for more information.
+"""
+
+
+class IncompleteConfiguration(UserWarning):
+    pass
+
 
 def _disable_insecure_serializers(whitelist=[]):
-    for name in set("pickle", "json", "yaml", "msgpack") - set(whitelist):
+    for name in set(registry._decoders.keys()) - set(whitelist):
         try:
-            unregister(name)
+            registry.disable(name)
         except SerializerNotInstalled:
             pass
 
 
-def setup_security(allowed_serializers=[]):
+def setup_security(allowed_serializers=[], key=None, cert=None, store=None,
+        digest=None, serializer=None):
     """setup secure serialization"""
     conf = current_app.conf
     if conf.CELERY_TASK_SERIALIZER != "auth":
-        return
+        return warn(IncompleteConfiguration(AUTH_DISABLED))
 
     try:
         from OpenSSL import crypto  # noqa
     except ImportError:
-        raise ImproperlyConfigured(
-            "You need to install the pyOpenSSL library to use "
-            "the auth serializer.")
+        raise ImproperlyConfigured(SSL_NOT_INSTALLED)
 
-    key = conf.CELERY_SECURITY_KEY
-    cert = conf.CELERY_SECURITY_CERTIFICATE
-    store = conf.CELERY_SECURITY_CERT_STORE
+    key = key or conf.CELERY_SECURITY_KEY
+    cert = cert or conf.CELERY_SECURITY_CERTIFICATE
+    store = store or conf.CELERY_SECURITY_CERT_STORE
+    digest = digest or conf.CELERY_SECURITY_DIGEST
+    serializer = serializer or conf.CELERY_SECURITY_SERIALIZER
 
-    if key is None or cert is None or store is None:
-        raise ImproperlyConfigured(
-            "CELERY_SECURITY_KEY, CELERY_SECURITY_CERTIFICATE and "
-            "CELERY_SECURITY_CERT_STORE options are required "
-            "settings when using the auth serializer")
+    if any(not v for v in (key, cert, store)):
+        raise ImproperlyConfigured(SETTING_MISSING)
 
+    _disable_insecure_serializers(allowed_serializers)
     with open(key) as kf:
         with open(cert) as cf:
             register_auth(kf.read(), cf.read(), store)
-    _disable_insecure_serializers(allowed_serializers)

celery/security/serialization.py

@@ -3,7 +3,7 @@ from __future__ import absolute_import
 import anyjson
 import base64
 
-from kombu.serialization import registry
+from kombu.serialization import registry, encode, decode
 
 from ..exceptions import SecurityError
 from ..utils.encoding import bytes_to_str, str_to_bytes
@@ -23,27 +23,35 @@ def b64decode(s):
 class SecureSerializer(object):
 
     def __init__(self, key=None, cert=None, cert_store=None,
-            serialize=anyjson.serialize,
-            deserialize=anyjson.deserialize,
-            digest='sha1'):
+            digest="sha1", serializer="json"):
         self._key = key
         self._cert = cert
         self._cert_store = cert_store
-        self._serialize = serialize
-        self._deserialize = deserialize
         self._digest = digest
+        self._serialize = anyjson.serialize
+        self._deserialize = anyjson.deserialize
+        self._serializer = serializer
 
     def serialize(self, data):
         """serialize data structure into string"""
         assert self._key is not None
         assert self._cert is not None
         try:
-            data = self._serialize(data)
-            signature = b64encode(self._key.sign(data, self._digest))
+            content_type, content_encoding, body = encode(
+                    data, serializer=self._serializer)
+            # What we sign is the serialized body, not the body itself.
+            # this way the receiver doesn't have to decode the contents
+            # to verify the signature (and thus avoiding potential flaws
+            # in the decoding step).
+            signature = b64encode(self._key.sign(body, self._digest))
             signer = self._cert.get_id()
-            return self._serialize(dict(data=data,
-                                        signer=signer,
-                                        signature=signature))
+            return self._serialize({
+                    "body": body,
+                    "signer": self._cert.get_id(),
+                    "signature": signature,
+                    "content_type": content_type,
+                    "content_encoding": content_encoding,
+            })
         except Exception, exc:
             raise SecurityError("Unable to serialize: %r" % (exc, ))
 
@@ -54,19 +62,22 @@ class SecureSerializer(object):
             data = self._deserialize(data)
             signature = b64decode(data["signature"])
             signer = data["signer"]
-            data = data["data"]
-            self._cert_store[signer].verify(data, signature, self._digest)
-            return self._deserialize(data)
+            body = data["body"]
+            self._cert_store[signer].verify(body,
+                                            signature, self._digest)
+            return decode(body, data["content_type"],
+                                data["content_encoding"], force=True)
         except Exception, exc:
             raise SecurityError("Unable to deserialize: %r" % (exc, ))
 
 
-def register_auth(key=None, cert=None, store=None):
+def register_auth(key=None, cert=None, store=None, digest="sha1",
+        serializer="pickle"):
     """register security serializer"""
     s = SecureSerializer(key and PrivateKey(key),
                          cert and Certificate(cert),
                          store and FSCertStore(store),
-                         anyjson.serialize, anyjson.deserialize)
+                         digest=digest, serializer=serializer)
     registry.register("auth", s.serialize, s.deserialize,
                       content_type="application/data",
                       content_encoding="utf-8")