Adds security package for message signing

celery/security/__init__.py

@@ -0,0 +1,32 @@
+from kombu.serialization import unregister, SerializerNotInstalled
+
+from celery.app import app_or_default
+from celery.security.serialization import register_auth
+from celery.exceptions import ImproperlyConfigured
+
+def _disable_insecure_serializers():
+    for name in ('pickle', 'json', 'yaml', 'msgpack'):
+        try:
+            unregister(name)
+        except SerializerNotInstalled:
+            pass
+
+def setup_security():
+    """setup secure serialization"""
+    conf = app_or_default().conf
+    if conf.CELERY_TASK_SERIALIZER != 'auth':
+        return
+
+    key = getattr(conf, 'CELERY_SECURITY_KEY', None)
+    cert = getattr(conf, 'CELERY_SECURITY_CERTIFICATE', None)
+    store = getattr(conf, 'CELERY_SECURITY_CERT_STORE', None)
+
+    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 "
+            "for auth serializer")
+
+    with open(key) as kf, open(cert) as cf:
+        register_auth(kf.read(), cf.read(), store)
+    _disable_insecure_serializers()

celery/security/certificate.py

@@ -0,0 +1,76 @@
+import os
+import glob
+
+from OpenSSL import crypto
+from OpenSSL.crypto import FILETYPE_PEM
+
+from celery.security.exceptions import SecurityError
+
+class Certificate(object):
+    """X.509 certificate"""
+    def __init__(self, cert):
+        try:
+            self._cert = crypto.load_certificate(FILETYPE_PEM, cert)
+        except crypto.Error, e:
+            raise SecurityError("Invalid certificate", e)
+
+    def has_expired(self):
+        """check if the certificate has expired"""
+        return self._cert.has_expired()
+
+    def get_serial_number(self):
+        """return certificate serial number"""
+        return self._cert.get_serial_number()
+
+    def get_issuer(self):
+        """return issuer (CA) as a string"""
+        return ' '.join(map(lambda x: x[1],
+                            self._cert.get_issuer().get_components()))
+
+    def get_id(self):
+        """serial number/issuer pair uniquely identifies a certificate"""
+        return "%s %s" % (self.get_issuer(), self.get_serial_number())
+
+    def verify(self, data, signature):
+        """verify the signature for a data string"""
+        try:
+            crypto.verify(self._cert, signature, data, 'sha1')
+        except crypto.Error, e:
+            raise SecurityError("Bad signature", e)
+
+class CertStore(object):
+    """Base class for certificate stores"""
+    def __init__(self):
+        self._certs = {}
+
+    def itercerts(self):
+        """an iterator over the certificates"""
+        for c in self._certs.itervalues():
+            yield c
+
+    def __getitem__(self, id):
+        """get certificate by id"""
+        try:
+            return self._certs[id]
+        except KeyError, e:
+            raise SecurityError("Unknown certificate: %s" % id, e)
+
+    def add_cert(self, cert):
+        if cert.get_id() in self._certs:
+            raise SecurityError("Duplicate certificate: %s" % id)
+        self._certs[cert.get_id()] = cert
+
+class FSCertStore(CertStore):
+    """File system certificate store"""
+    def __init__(self, path):
+        CertStore.__init__(self)
+        if os.path.isdir(path):
+            path = os.path.join(path, '*')
+        for p in glob.glob(path):
+            with open(p) as f:
+                cert = Certificate(f.read())
+                if cert.has_expired():
+                    raise SecurityError("Expired certificate: %s" %\
+                                        cert.get_id())
+                self.add_cert(cert)
+

celery/security/exceptions.py

@@ -0,0 +1,5 @@
+class SecurityError(Exception):
+    """Security related exceptions"""
+    def __init__(self, msg, exc=None, *args, **kwargs):
+        Exception.__init__(self, msg, exc, *args, **kwargs)
+

celery/security/key.py

@@ -0,0 +1,19 @@
+from OpenSSL import crypto
+from OpenSSL.crypto import FILETYPE_PEM
+
+from celery.security.exceptions import SecurityError
+
+class PrivateKey(object):
+    def __init__(self, key):
+        try:
+            self._key = crypto.load_privatekey(FILETYPE_PEM, key)
+        except crypto.Error, e:
+            raise SecurityError("Invalid private key", e)
+
+    def sign(self, data):
+        """sign a data string"""
+        try:
+            return crypto.sign(self._key, data, 'sha1')
+        except crypto.Error, e:
+            raise SecurityError("Unable to sign a data string", e)
+

celery/security/serialization.py

@@ -0,0 +1,45 @@
+import pickle
+
+from kombu.serialization import registry
+
+from certificate import Certificate, FSCertStore
+from key import PrivateKey
+
+class SecureSerializer(object):
+
+    def __init__(self, key=None, cert=None, cert_store=None):
+        self._key = key
+        self._cert = cert
+        self._cert_store = cert_store
+
+    def serialize(self, data):
+        """serialize data structure into string"""
+        assert self._key is not None
+        assert self._cert is not None
+        data = pickle.dumps(data)
+        signature = self._key.sign(data)
+        signer = self._cert.get_id()
+        return pickle.dumps(dict(data=data,
+                                 signer=signer,
+                                 signature=signature))
+
+    def deserialize(self, data):
+        """deserialize data structure from string"""
+        assert self._cert_store is not None
+        data = pickle.loads(data)
+        signature = data['signature']
+        signer = data['signer']
+        data = data['data']
+        self._cert_store[signer].verify(data, signature)
+        return pickle.loads(data)
+
+def register_auth(key=None, cert=None, store=None):
+    """register security serializer"""
+    global s
+    s = SecureSerializer(key and PrivateKey(key),
+                         cert and Certificate(cert),
+                         store and FSCertStore(store))
+    registry.register("auth", s.serialize, s.deserialize,
+                      content_type='application/data',
+                      content_encoding='binary')
+