Fixed redirect testing method, and some minor changes to cache and storage

server/src/tests/core/util/middleware/test_redirect.py

@@ -45,7 +45,7 @@ logger = logging.getLogger(__name__)
 
 class RedirectMiddlewareTest(test.UDSTransactionTestCase):
     """
-    Test actor functionality
+    Test client functionality
     """
     def test_redirect(self):
         RedirectMiddlewareTest.add_middleware('uds.core.util.middleware.redirect.RedirectMiddleware')
@@ -70,11 +70,11 @@ class RedirectMiddlewareTest(test.UDSTransactionTestCase):
         # These urls will never redirect:
         for url in _NO_REDIRECT:
             # Append some random string to avoid cache and make a 404 occur
-            url = f'/{url}/{CryptoManager().randomString(32)}'
+            url = f'/{url}{("/" + CryptoManager().randomString(32)) if "test" not in url else ""}'
             response = self.client.get(url, secure=False)
-            # every url will return 404, except /uds/rest/client that will return 400 and wyse or servlet that will return 302
-            if url.startswith('/uds/rest/client'):
-                self.assertEqual(response.status_code, 400)
+            # every url will return 404, except /uds/rest/client/test that will return 200 and wyse or servlet that will return 302
+            if url.startswith('/uds/rest/client/test'):
+                self.assertEqual(response.status_code, 200)
             elif url.startswith('/wyse') or url.startswith('/servlet'):
                 self.assertEqual(response.status_code, 302)
             else:

server/src/uds/__init__.py

@@ -95,7 +95,7 @@ class UDSAppConfig(AppConfig):
 default_app_config = 'uds.UDSAppConfig'
 
 
-# Sets up several sqlite non existing methods
+# Sets up several sqlite non existing methodsm and some optimizations on sqlite
 @receiver(connection_created)
 def extend_sqlite(connection=None, **kwargs):
     if connection and connection.vendor == "sqlite":

server/src/uds/core/util/cache.py

@@ -31,7 +31,7 @@
 """
 import datetime
 import codecs
-import pickle  # nosec: This is e controled pickle loading
+import yaml
 import typing
 import logging
 
@@ -58,6 +58,19 @@ class Cache:
     _owner: str
     _bowner: bytes
 
+    @staticmethod
+    def _basic_serialize(value: typing.Any) -> str:
+        return codecs.encode(yaml.safe_dump(value).encode(), 'base64').decode()
+
+    @staticmethod
+    def _basic_deserialize(value: str) -> typing.Any:
+        return yaml.safe_load(codecs.decode(value.encode(), 'base64'))
+
+    _serializer: typing.ClassVar[typing.Callable[[typing.Any], str]] = _basic_serialize
+    _deserializer: typing.ClassVar[
+        typing.Callable[[str], typing.Any]
+    ] = _basic_deserialize
+
     def __init__(self, owner: typing.Union[str, bytes]):
         self._owner = owner.decode('utf-8') if isinstance(owner, bytes) else owner
         self._bowner = self._owner.encode('utf8')
@@ -82,11 +95,9 @@ class Cache:
 
             try:
                 # logger.debug('value: %s', c.value)
-                val = pickle.loads(  # nosec: This is e controled pickle loading
-                    typing.cast(bytes, codecs.decode(c.value.encode(), 'base64'))
-                )
+                val = Cache._deserializer(c.value)
             except Exception:  # If invalid, simple do no tuse it
-                logger.exception('Invalid pickle from cache. Removing it.')
+                # logger.exception('Invalid yaml from cache. Removing it.')
                 c.delete()
                 return defValue
 
@@ -141,7 +152,7 @@ class Cache:
         if validity is None:
             validity = Cache.DEFAULT_VALIDITY
         key = self.__getKey(skey)
-        strValue: str = codecs.encode(pickle.dumps(value), 'base64').decode()
+        strValue = Cache._serializer(value)
         now = getSqlDatetime()
         try:
             DBCache.objects.create(

server/src/uds/core/util/storage.py

@@ -375,7 +375,7 @@ class Storage:
 
     def filter(
         self, attr1: typing.Optional[str] = None, forUpdate: bool = False
-    ) -> typing.Iterable[typing.Tuple[str, bytes, str]]:
+    ) -> typing.Iterable[typing.Tuple[str, bytes, 'str|None']]:
         if attr1 is None:
             query = DBStorage.objects.filter(owner=self._owner)  # @UndefinedVariable
         else:
@@ -391,7 +391,7 @@ class Storage:
 
     def filterPickle(
         self, attr1: typing.Optional[str] = None, forUpdate: bool = False
-    ) -> typing.Iterable[typing.Tuple[str, typing.Any, str]]:
+    ) -> typing.Iterable[typing.Tuple[str, typing.Any, 'str|None']]:
         for v in self.filter(attr1, forUpdate):
             yield (v[0], pickle.loads(v[1]), v[2])  # nosec: secure pickle load