Backport of security from 3.0-devel

celery/exceptions.py

@@ -16,6 +16,14 @@ Task of kind %s is not registered, please make sure it's imported.\
 """
 
 
+class SecurityError(Exception):
+    """Security related exceptions.
+
+    Handle with care.
+
+    """
+
+
 class SystemTerminate(SystemExit):
     """Signals that the worker should terminate."""
 

celery/security/__init__.py

@@ -0,0 +1,46 @@
+from __future__ import absolute_import
+from __future__ import with_statement
+
+from kombu.serialization import unregister, SerializerNotInstalled
+
+from .. import current_app
+from ..exceptions import ImproperlyConfigured
+
+from .serialization import register_auth
+
+
+def _disable_insecure_serializers():
+    for name in ("pickle", "json", "yaml", "msgpack"):
+        try:
+            unregister(name)
+        except SerializerNotInstalled:
+            pass
+
+
+def setup_security():
+    """setup secure serialization"""
+    conf = current_app.conf
+    if conf.CELERY_TASK_SERIALIZER != "auth":
+        return
+
+    try:
+        from OpenSSL import crypto  # noqa
+    except ImportError:
+        raise ImproperlyConfigured(
+            "You need to install the pyOpenSSL library to use "
+            "the auth serializer.")
+
+    key = conf.CELERY_SECURITY_KEY
+    cert = conf.CELERY_SECURITY_CERTIFICATE
+    store = conf.CELERY_SECURITY_CERT_STORE
+
+    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")
+
+    with open(key) as kf:
+        with open(cert) as cf:
+            register_auth(kf.read(), cf.read(), store)
+    _disable_insecure_serializers()

celery/security/certificate.py

@@ -0,0 +1,87 @@
+from __future__ import absolute_import
+from __future__ import with_statement
+
+import os
+import glob
+
+try:
+    from OpenSSL import crypto
+except ImportError:
+    crypto = None  # noqa
+
+from ..exceptions import SecurityError
+
+
+class Certificate(object):
+    """X.509 certificate."""
+
+    def __init__(self, cert):
+        assert crypto is not None
+        try:
+            self._cert = crypto.load_certificate(crypto.FILETYPE_PEM, cert)
+        except crypto.Error, exc:
+            raise SecurityError("Invalid certificate: %r" % (exc, ))
+
+    def has_expired(self):
+        """Check if the certificate has expired."""
+        return self._cert.has_expired()
+
+    def get_serial_number(self):
+        """Returns the certificates serial number."""
+        return self._cert.get_serial_number()
+
+    def get_issuer(self):
+        """Returns issuer (CA) as a string"""
+        return ' '.join(x[1] for x in
+                        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):
+        """Verifies the signature for string containing data."""
+        try:
+            crypto.verify(self._cert, signature, data, 'sha1')
+        except crypto.Error, exc:
+            raise SecurityError("Bad signature: %r" % (exc, ))
+
+
+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:
+            raise SecurityError("Unknown certificate: %r" % (id, ))
+
+    def add_cert(self, cert):
+        if cert.get_id() in self._certs:
+            raise SecurityError("Duplicate certificate: %r" % (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: %r" % (cert.get_id(), ))
+                self.add_cert(cert)

celery/security/key.py

@@ -0,0 +1,25 @@
+from __future__ import absolute_import
+
+try:
+    from OpenSSL import crypto
+except ImportError:
+    crypto = None  # noqa
+
+from ..exceptions import SecurityError
+
+
+class PrivateKey(object):
+
+    def __init__(self, key):
+        assert crypto is not None
+        try:
+            self._key = crypto.load_privatekey(crypto.FILETYPE_PEM, key)
+        except crypto.Error, exc:
+            raise SecurityError("Invalid private key: %r" % (exc, ))
+
+    def sign(self, data):
+        """sign string containing data."""
+        try:
+            return crypto.sign(self._key, data, "sha1")
+        except crypto.Error, exc:
+            raise SecurityError("Unable to sign data: %r" % (exc, ))

celery/security/serialization.py

@@ -0,0 +1,63 @@
+from __future__ import absolute_import
+
+from base64 import b64encode, b64decode
+
+import anyjson
+import base64
+
+from kombu.serialization import registry
+
+from ..exceptions import SecurityError
+
+from .certificate import Certificate, FSCertStore
+from .key import PrivateKey
+
+
+class SecureSerializer(object):
+
+    def __init__(self, key=None, cert=None, cert_store=None,
+                       serialize=anyjson.serialize,
+                       deserialize=anyjson.deserialize):
+        self._key = key
+        self._cert = cert
+        self._cert_store = cert_store
+        self._serialize = serialize
+        self._deserialize = deserialize
+
+    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))
+            signer = self._cert.get_id()
+            return self._serialize(dict(data=data,
+                                        signer=signer,
+                                        signature=signature))
+        except Exception, exc:
+            raise SecurityError("Unable to serialize: %r" % (exc, ))
+
+    def deserialize(self, data):
+        """deserialize data structure from string"""
+        assert self._cert_store is not None
+        try:
+            data = self._deserialize(data)
+            signature = b64decode(data["signature"])
+            signer = data["signer"]
+            data = data["data"]
+            self._cert_store[signer].verify(data, signature)
+            return self._deserialize(data)
+        except Exception, exc:
+            raise SecurityError("Unable to deserialize: %r" % (exc, ))
+
+
+def register_auth(key=None, cert=None, store=None):
+    """register security serializer"""
+    s = SecureSerializer(key and PrivateKey(key),
+                         cert and Certificate(cert),
+                         store and FSCertStore(store),
+                         anyjson.serialize, anyjson.deserialize)
+    registry.register("auth", s.serialize, s.deserialize,
+                      content_type="application/data",
+                      content_encoding="utf-8")

celery/tests/test_security/__init__.py

@@ -0,0 +1,75 @@
+from __future__ import absolute_import
+"""
+Keys and certificates for tests (KEY1 is a private key of CERT1, etc.)
+
+Generated with::
+
+    $ openssl genrsa -des3 -passout pass:test -out key1.key 1024
+    $ openssl req -new -key key1.key -out key1.csr -passin pass:test
+    $ cp key1.key key1.key.org
+    $ openssl rsa -in key1.key.org -out key1.key -passin pass:test
+    $ openssl x509 -req -days 365 -in cert1.csr \
+              -signkey key1.key -out cert1.crt
+    $ rm key1.key.org cert1.csr
+
+"""
+
+KEY1 = """-----BEGIN RSA PRIVATE KEY-----
+MIICXgIBAAKBgQDCsmLC+eqL4z6bhtv0nzbcnNXuQrZUoh827jGfDI3kxNZ2LbEy
+kJOn7GIl2tPpcY2Dm1sOM8G1XLm/8Izprp4ifpF4Gi0mqz0GquY5dcMNASG9zkRO
+J1z8dQUyp3PIUHdQdrKbYQVifkA4dh6Kg27k8/IcdY1lHsaIju4bX7MADwIDAQAB
+AoGBAKWpCRWjdiluwu+skO0Up6aRIAop42AhzfN8OuZ81SMJRP2rJTHECI8COATD
+rDneb63Ce3ibG0BI1Jf3gr624D806xVqK/SVHZNbfWx0daE3Q43DDk1UdhRF5+0X
+HPqqU/IdeW1YGyWJi+IhMTXyGqhZ1BTN+4vHL7NlRpDt6JOpAkEA+xvfRO4Ca7Lw
+NEgvW7n+/L9b+xygQBtOA5s260pO+8jMrXvOdCjISaKHD8HZGFN9oUmLsDXXBhjh
+j0WCMdsHbQJBAMZ9OIw6M/Uxv5ANPCD58p6PZTb0knXVPMYBFQ7Y/h2HZzqbEyiI
+DLGZpAa9/IhVkoCULd/TNytz5rl27KEni+sCQArFQEdZmhr6etkTO4zIpoo6vvw/
+VxRI14jKEIn5Dvg3vae3RryuvyCBax+e5evoMNxJJkexl354dLxLc/ElfuUCQQCq
+U14pBvD7ITuPM6w7aAEIi2iBZhIgR2GlT9xwJ0i4si6lHdms2EJ8TKlyl6mSnEvh
+RkavYSJgiU6eLC0WhUcNAkEA7vuNcz/uuckmq870qfSzUQJIYLzwVOadEdEEAVy0
+L0usztlKmAH8U/ceQMMJLMI9W4m680JrMf3iS7f+SkgUTA==
+-----END RSA PRIVATE KEY-----"""
+
+KEY2 = """-----BEGIN RSA PRIVATE KEY-----
+MIICXQIBAAKBgQDH22L8b9AmST9ABDmQTQ2DWMdDmK5YXZt4AIY81IcsTQ/ccM0C
+fwXEP9tdkYwtcxMCWdASwY5pfMy9vFp0hyrRQMSNfuoxAgONuNWPyQoIvY3ZXRe6
+rS+hb/LN4+vdjX+oxmYiQ2HmSB9rh2bepE6Cw+RLJr5sXXq+xZJ+BLt5tQIDAQAB
+AoGBAMGBO0Arip/nP6Rd8tYypKjN5nEefX/1cjgoWdC//fj4zCil1vlZv12abm0U
+JWNEDd2y0/G1Eow0V5BFtFcrIFowU44LZEiSf7sKXlNHRHlbZmDgNXFZOt7nVbHn
+6SN+oCYjaPjji8idYeb3VQXPtqMoMn73MuyxD3k3tWmVLonpAkEA6hsu62qhUk5k
+Nt88UZOauU1YizxsWvT0bHioaceE4TEsbO3NZs7dmdJIcRFcU787lANaaIq7Rw26
+qcumME9XhwJBANqMOzsYQ6BX54UzS6x99Jjlq9MEbTCbAEZr/yjopb9f617SwfuE
+AEKnIq3HL6/Tnhv3V8Zy3wYHgDoGNeTVe+MCQQDi/nyeNAQ8RFqTgh2Ak/jAmCi0
+yV/fSgj+bHgQKS/FEuMas/IoL4lbrzQivkyhv5lLSX0ORQaWPM+z+A0qZqRdAkBh
+XE+Wx/x4ljCh+nQf6AzrgIXHgBVUrfi1Zq9Jfjs4wnaMy793WRr0lpiwaigoYFHz
+i4Ei+1G30eeh8dpYk3KZAkB0ucTOsQynDlL5rLGYZ+IcfSfH3w2l5EszY47kKQG9
+Fxeq/HOp9JYw4gRu6Ycvqu57KHwpHhR0FCXRBxuYcJ5V
+-----END RSA PRIVATE KEY-----"""
+
+CERT1 = """-----BEGIN CERTIFICATE-----
+MIICATCCAWoCCQCR6B3XQcBOvjANBgkqhkiG9w0BAQUFADBFMQswCQYDVQQGEwJB
+VTETMBEGA1UECAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50ZXJuZXQgV2lkZ2l0
+cyBQdHkgTHRkMB4XDTExMDcxOTA5MDgyMloXDTEyMDcxODA5MDgyMlowRTELMAkG
+A1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoMGEludGVybmV0
+IFdpZGdpdHMgUHR5IEx0ZDCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEAwrJi
+wvnqi+M+m4bb9J823JzV7kK2VKIfNu4xnwyN5MTWdi2xMpCTp+xiJdrT6XGNg5tb
+DjPBtVy5v/CM6a6eIn6ReBotJqs9BqrmOXXDDQEhvc5ETidc/HUFMqdzyFB3UHay
+m2EFYn5AOHYeioNu5PPyHHWNZR7GiI7uG1+zAA8CAwEAATANBgkqhkiG9w0BAQUF
+AAOBgQA4+OiJ+pyq9lbEMFYC9K2+e77noHJkwUOs4wO6p1R14ZqSmoIszQ7KEBiH
+2HHPMUY6kt4GL1aX4Vr1pUlXXdH5WaEk0fvDYZemILDMqIQJ9ettx8KihZjFGC4k
+Y4Sy5xmqdE9Kjjd854gTRRnzpMnJp6+74Ki2X8GHxn3YBM+9Ng==
+-----END CERTIFICATE-----"""
+
+CERT2 = """-----BEGIN CERTIFICATE-----
+MIICATCCAWoCCQCV/9A2ZBM37TANBgkqhkiG9w0BAQUFADBFMQswCQYDVQQGEwJB
+VTETMBEGA1UECAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50ZXJuZXQgV2lkZ2l0
+cyBQdHkgTHRkMB4XDTExMDcxOTA5MDkwMloXDTEyMDcxODA5MDkwMlowRTELMAkG
+A1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoMGEludGVybmV0
+IFdpZGdpdHMgUHR5IEx0ZDCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEAx9ti
+/G/QJkk/QAQ5kE0Ng1jHQ5iuWF2beACGPNSHLE0P3HDNAn8FxD/bXZGMLXMTAlnQ
+EsGOaXzMvbxadIcq0UDEjX7qMQIDjbjVj8kKCL2N2V0Xuq0voW/yzePr3Y1/qMZm
+IkNh5kgfa4dm3qROgsPkSya+bF16vsWSfgS7ebUCAwEAATANBgkqhkiG9w0BAQUF
+AAOBgQBzaZ5vBkzksPhnWb2oobuy6Ne/LMEtdQ//qeVY4sKl2tOJUCSdWRen9fqP
+e+zYdEdkFCd8rp568Eiwkq/553uy4rlE927/AEqs/+KGYmAtibk/9vmi+/+iZXyS
+WWZybzzDZFncq1/N1C3Y/hrCBNDFO4TsnTLAhWtZ4c0vDAiacw==
+-----END CERTIFICATE-----"""

celery/tests/test_security/case.py

@@ -0,0 +1,14 @@
+from __future__ import absolute_import
+
+from nose import SkipTest
+
+from celery.tests.utils import unittest
+
+
+class SecurityCase(unittest.TestCase):
+
+    def setUp(self):
+        try:
+            from OpenSSL import crypto  # noqa
+        except ImportError:
+            raise SkipTest("OpenSSL.crypto not installed")

celery/tests/test_security/test_certificate.py

@@ -0,0 +1,41 @@
+from __future__ import absolute_import
+
+from celery.exceptions import SecurityError
+from celery.security.certificate import Certificate, CertStore
+
+from . import CERT1, CERT2, KEY1
+from .case import SecurityCase
+
+
+class TestCertificate(SecurityCase):
+
+    def test_valid_certificate(self):
+        Certificate(CERT1)
+        Certificate(CERT2)
+
+    def test_invalid_certificate(self):
+        self.assertRaises(TypeError, Certificate, None)
+        self.assertRaises(SecurityError, Certificate, "")
+        self.assertRaises(SecurityError, Certificate, "foo")
+        self.assertRaises(SecurityError, Certificate, CERT1[:20] + CERT1[21:])
+        self.assertRaises(SecurityError, Certificate, KEY1)
+
+
+class TestCertStore(SecurityCase):
+
+    def test_itercerts(self):
+        cert1 = Certificate(CERT1)
+        cert2 = Certificate(CERT2)
+        certstore = CertStore()
+        for c in certstore.itercerts():
+            self.assertTrue(False)
+        certstore.add_cert(cert1)
+        certstore.add_cert(cert2)
+        for c in certstore.itercerts():
+            self.assertIn(c, (cert1, cert2))
+
+    def test_duplicate(self):
+        cert1 = Certificate(CERT1)
+        certstore = CertStore()
+        certstore.add_cert(cert1)
+        self.assertRaises(SecurityError, certstore.add_cert, cert1)

celery/tests/test_security/test_key.py

@@ -0,0 +1,21 @@
+from __future__ import absolute_import
+
+from celery.exceptions import SecurityError
+from celery.security.key import PrivateKey
+
+from . import CERT1, KEY1, KEY2
+from .case import SecurityCase
+
+
+class TestKey(SecurityCase):
+
+    def test_valid_private_key(self):
+        PrivateKey(KEY1)
+        PrivateKey(KEY2)
+
+    def test_invalid_private_key(self):
+        self.assertRaises(TypeError, PrivateKey, None)
+        self.assertRaises(SecurityError, PrivateKey, "")
+        self.assertRaises(SecurityError, PrivateKey, "foo")
+        self.assertRaises(SecurityError, PrivateKey, KEY1[:20] + KEY1[21:])
+        self.assertRaises(SecurityError, PrivateKey, CERT1)

celery/tests/test_security/test_serialization.py

@@ -0,0 +1,50 @@
+from __future__ import absolute_import
+
+from celery.exceptions import SecurityError
+
+from celery.security.serialization import SecureSerializer
+from celery.security.certificate import Certificate, CertStore
+from celery.security.key import PrivateKey
+
+from . import CERT1, CERT2, KEY1, KEY2
+from .case import SecurityCase
+
+
+class TestSecureSerializer(SecurityCase):
+
+    def _get_s(self, key, cert, certs):
+        store = CertStore()
+        for c in certs:
+            store.add_cert(Certificate(c))
+        return SecureSerializer(PrivateKey(key), Certificate(cert), store)
+
+    def test_serialize(self):
+        s = self._get_s(KEY1, CERT1, [CERT1])
+        self.assertEqual(s.deserialize(s.serialize("foo")), "foo")
+
+    def test_deserialize(self):
+        s = self._get_s(KEY1, CERT1, [CERT1])
+        self.assertRaises(SecurityError, s.deserialize, "bad data")
+
+    def test_unmatched_key_cert(self):
+        s = self._get_s(KEY1, CERT2, [CERT1, CERT2])
+        self.assertRaises(SecurityError,
+                          s.deserialize, s.serialize("foo"))
+
+    def test_unknown_source(self):
+        s1 = self._get_s(KEY1, CERT1, [CERT2])
+        s2 = self._get_s(KEY1, CERT1, [])
+        self.assertRaises(SecurityError,
+                          s1.deserialize, s1.serialize("foo"))
+        self.assertRaises(SecurityError,
+                          s2.deserialize, s2.serialize("foo"))
+
+    def test_self_send(self):
+        s1 = self._get_s(KEY1, CERT1, [CERT1])
+        s2 = self._get_s(KEY1, CERT1, [CERT1])
+        self.assertEqual(s2.deserialize(s1.serialize("foo")), "foo")
+
+    def test_separate_ends(self):
+        s1 = self._get_s(KEY1, CERT1, [CERT2])
+        s2 = self._get_s(KEY2, CERT2, [CERT1])
+        self.assertEqual(s2.deserialize(s1.serialize("foo")), "foo")

requirements/security.txt

@@ -0,0 +1 @@
+PyOpenSSL

requirements/test.txt

@@ -7,3 +7,4 @@ mock>=0.7.0
 redis
 pymongo
 SQLAlchemy
+PyOpenSSL