SecureSerializer: export DEFAULT_SEPARATOR to const

celery/security/serialization.py

@@ -11,6 +11,11 @@
 
 __all__ = ('SecureSerializer', 'register_auth')
 
+# Note: we guarantee that this value won't appear in the serialized data,
+# so we can use it as a separator.
+# If you change this value, make sure it's not present in the serialized data.
+DEFAULT_SEPARATOR = str_to_bytes("\x00\x01")
+
 
 class SecureSerializer:
     """Signed serializer."""
@@ -53,14 +58,14 @@ def deserialize(self, data):
                      payload['content_encoding'], force=True)
 
     def _pack(self, body, content_type, content_encoding, signer, signature,
-              sep=str_to_bytes('\x00\x01')):
+              sep=DEFAULT_SEPARATOR):
         fields = sep.join(
             ensure_bytes(s) for s in [b64encode(signer), b64encode(signature),
                                       content_type, content_encoding, body]
         )
         return b64encode(fields)
 
-    def _unpack(self, payload, sep=str_to_bytes('\x00\x01')):
+    def _unpack(self, payload, sep=DEFAULT_SEPARATOR):
         raw_payload = b64decode(ensure_bytes(payload))
         v = raw_payload.split(sep, maxsplit=4)
         return {

t/unit/security/test_serialization.py

@@ -8,7 +8,7 @@
 from celery.exceptions import SecurityError
 from celery.security.certificate import Certificate, CertStore
 from celery.security.key import PrivateKey
-from celery.security.serialization import SecureSerializer, register_auth
+from celery.security.serialization import DEFAULT_SEPARATOR, SecureSerializer, register_auth
 
 from . import CERT1, CERT2, KEY1, KEY2
 from .case import SecurityCase
@@ -24,7 +24,9 @@ def _get_s(self, key, cert, certs, serializer="json"):
             PrivateKey(key), Certificate(cert), store, serializer=serializer
         )
 
-    @pytest.mark.parametrize("data", [1, "foo", b"foo", {"foo": 1}, {"foo": "\x00\x01"}])
+    @pytest.mark.parametrize(
+        "data", [1, "foo", b"foo", {"foo": 1}, {"foo": DEFAULT_SEPARATOR}]
+    )
     @pytest.mark.parametrize("serializer", ["json", "pickle"])
     def test_serialize(self, data, serializer):
         s = self._get_s(KEY1, CERT1, [CERT1], serializer=serializer)