Fixing up bandit recomendations & minor typo errors

server/src/uds/auths/InternalDB/authenticator.py

@@ -93,23 +93,23 @@ class InternalDBAuth(auths.Authenticator):
         tab=gui.ADVANCED_TAB,
     )
 
-    def getIp(self) -> str:
+    def getIp(self, request: 'ExtendedHttpRequest') -> str:
         ip = (
-            getRequest().ip_proxy if self.acceptProxy.isTrue() else getRequest().ip
+            request.ip_proxy if self.acceptProxy.isTrue() else request.ip
         )  # pylint: disable=maybe-no-member
         if self.reverseDns.isTrue():
             try:
                 return str(
                     dns.resolver.query(dns.reversename.from_address(ip).to_text(), 'PTR')[0]
                 )
-            except Exception:
+            except Exception:  # nosec: intentionally
                 pass
         return ip
 
     def mfaIdentifier(self, username: str) -> str:
         try:
             self.dbAuthenticator().users.get(name=username, state=State.ACTIVE).mfaData
-        except Exception:
+        except Exception:  # nosec: This is e controled pickle loading
             pass
         return ''
 
@@ -134,7 +134,7 @@ class InternalDBAuth(auths.Authenticator):
                 usr.name = newUsername
                 usr.parent = parent
                 usr.save()
-            except Exception:
+            except Exception:  # nosec: intentionally
                 pass  # User already exists
             username = newUsername
 

server/src/uds/auths/SAML/saml.py

@@ -32,7 +32,7 @@
 """
 import re
 from urllib.parse import urlparse
-import xml.sax
+import xml.sax  # nosec: used to parse trusted xml provided only by administrators
 import datetime
 import requests
 import logging
@@ -204,6 +204,13 @@ class SAMLAuthenticator(auths.Authenticator):
         tab=gui.ADVANCED_TAB,
     )
 
+    checkSSLCertificate = gui.CheckBoxField(
+        label=_('Check SSL certificate'),
+        defvalue=False,  # For compatibility with previous versions
+        order=23,
+        tooltip=_('If set, check SSL certificate on requests for IDP Metadata'),
+        tab=_('Security'),
+    )
 
     nameIdEncrypted = gui.CheckBoxField(
         label=_('Encripted nameID'),
@@ -375,7 +382,7 @@ class SAMLAuthenticator(auths.Authenticator):
         if idpMetadata.startswith('http://') or idpMetadata.startswith('https://'):
             logger.debug('idp Metadata is an URL: %s', idpMetadata)
             try:
-                resp = requests.get(idpMetadata.split('\n')[0], verify=False)
+                resp = requests.get(idpMetadata.split('\n')[0], verify=self.checkSSLCertificate.isTrue())
                 idpMetadata = resp.content.decode()
             except Exception as e:
                 raise auths.Authenticator.ValidationException(
@@ -388,7 +395,7 @@ class SAMLAuthenticator(auths.Authenticator):
         # Try to parse it so we can check it is valid. Right now, it checks just that this is XML, will
         # correct it to check that is is valid idp metadata
         try:
-            xml.sax.parseString(idpMetadata, xml.sax.ContentHandler())  # type: ignore
+            xml.sax.parseString(idpMetadata, xml.sax.ContentHandler())  # type: ignore  # nosec: url provided by admin
         except Exception as e:
             msg = (gettext(' (obtained from URL)') if fromUrl else '') + str(e)
             raise auths.Authenticator.ValidationException(
@@ -439,7 +446,7 @@ class SAMLAuthenticator(auths.Authenticator):
     def getIdpMetadataDict(self, **kwargs) -> typing.Dict[str, typing.Any]:
         if self.idpMetadata.value.startswith('http'):
             try:
-                resp = requests.get(self.idpMetadata.value.split('\n')[0], verify=False)
+                resp = requests.get(self.idpMetadata.value.split('\n')[0], verify=self.checkSSLCertificate.isTrue())
                 val = resp.content.decode()
             except Exception as e:
                 logger.error('Error fetching idp metadata: %s', e)

server/src/uds/core/auths/auth.py

@@ -67,7 +67,7 @@ logger = logging.getLogger(__name__)
 authLogger = logging.getLogger('authLog')
 
 USER_KEY = 'uk'
-PASS_KEY = 'pk'
+PASS_KEY = 'pk'  # nosec: this is not a password but a cookie to store encrypted data
 EXPIRY_KEY = 'ek'
 AUTHORIZED_KEY = 'ak'
 ROOT_ID = -20091204  # Any negative number will do the trick
@@ -456,7 +456,9 @@ def webLogout(
         if request.user:
             authenticator = request.user.manager.getInstance()
             username = request.user.name
-            exit_url = authenticator.logout(username) or exit_url
+            logout = authenticator.logout(request, username)
+            if logout and logout.success == auths.AuthenticationSuccess.REDIRECT:
+                exit_url = logout.url
             if request.user.id != ROOT_ID:
                 # Log the event if not root user
                 events.addEvent(
@@ -524,7 +526,7 @@ def authLogLogin(
             ),
             log.WEB,
         )
-    except Exception:
+    except Exception:  # nosec: intentionally ignore exception
         pass
 
 

server/src/uds/core/environment.py

@@ -32,9 +32,12 @@
 """
 import typing
 
-from uds.core.util.cache import Cache
+
-from uds.core.util.storage import Storage
+if typing.TYPE_CHECKING:
-from uds.core.util.unique_id_generator import UniqueIDGenerator
+    from uds.core.util.cache import Cache
+    from uds.core.util.storage import Storage
+    from uds.core.util.unique_id_generator import UniqueIDGenerator
+
 
 TEMP_ENV = 'temporary'
 GLOBAL_ENV = 'global'
@@ -50,14 +53,14 @@ class Environment:
     __slots__ = ['_key', '_cache', '_storage', '_idGenerators']
 
     _key: str
-    _cache: Cache
+    _cache: 'Cache'
-    _storage: Storage
+    _storage: 'Storage'
-    _idGenerators: typing.Dict[str, UniqueIDGenerator]
+    _idGenerators: typing.Dict[str, 'UniqueIDGenerator']
 
     def __init__(
         self,
         uniqueKey: str,
-        idGenerators: typing.Optional[typing.Dict[str, UniqueIDGenerator]] = None,
+        idGenerators: typing.Optional[typing.Dict[str, 'UniqueIDGenerator']] = None,
     ):
         """
         Initialized the Environment for the specified id
@@ -66,6 +69,10 @@ class Environment:
             is used basically at User Services to auto-create ids for macs or names, using
             {'mac' : UniqueMacGenerator, 'name' : UniqueNameGenerator } as argument.
         """
+        # Avoid circular imports
+        from uds.core.util.cache import Cache
+        from uds.core.util.storage import Storage
+
         if idGenerators is None:
             idGenerators = dict()
         self._key = uniqueKey
@@ -74,7 +81,7 @@ class Environment:
         self._idGenerators = idGenerators
 
     @property
-    def cache(self) -> Cache:
+    def cache(self) -> 'Cache':
         """
         Method to acces the cache of the environment.
         @return: a referente to a Cache instance
@@ -82,14 +89,14 @@ class Environment:
         return self._cache
 
     @property
-    def storage(self) -> Storage:
+    def storage(self) -> 'Storage':
         """
         Method to acces the cache of the environment.
         @return: a referente to an Storage Instance
         """
         return self._storage
 
-    def idGenerators(self, generatorId: str) -> UniqueIDGenerator:
+    def idGenerators(self, generatorId: str) -> 'UniqueIDGenerator':
         """
         The idea of generator of id is to obtain at some moment Ids with a proper generator.
         If the environment do not contains generators of id, this method will return None.
@@ -112,8 +119,8 @@ class Environment:
         """
         Removes all related information from database for this environment.
         """
-        Cache.delete(self._key)
+        self._cache.clear()
-        Storage.delete(self._key)
+        self._storage.clear()
         for _, v in self._idGenerators.items():
             v.release()
 
@@ -156,8 +163,8 @@ class Environment:
         It will not make environment persistent
         """
         env = Environment(TEMP_ENV)
-        env.storage.clean()
+        env.storage.clear()
-        env.cache.clean()
+        env.cache.clear()
         return env
 
     @staticmethod
@@ -178,7 +185,7 @@ class Environmentable:
 
     _env: Environment
 
-    def __init__(self, environment: Environment):
+    def __init__(self, environment: 'Environment'):
         """
         Initialized the element
 
@@ -188,7 +195,7 @@ class Environmentable:
         self._env = environment
 
     @property
-    def env(self) -> Environment:
+    def env(self) -> 'Environment':
         """
         Utility method to access the envionment contained by this object
 
@@ -198,7 +205,7 @@ class Environmentable:
         return self._env
 
     @env.setter
-    def env(self, environment: Environment):
+    def env(self, environment: 'Environment'):
         """
         Assigns a new environment
 
@@ -208,7 +215,7 @@ class Environmentable:
         self._env = environment
 
     @property
-    def cache(self) -> Cache:
+    def cache(self) -> 'Cache':
         """
         Utility method to access the cache of the environment containe by this object
 
@@ -219,7 +226,7 @@ class Environmentable:
         return self._env.cache
 
     @property
-    def storage(self) -> Storage:
+    def storage(self) -> 'Storage':
         """
         Utility method to access the storage of the environment containe by this object
 
@@ -230,7 +237,7 @@ class Environmentable:
         """
         return self._env.storage
 
-    def idGenerators(self, generatorId: str) -> UniqueIDGenerator:
+    def idGenerators(self, generatorId: str) -> 'UniqueIDGenerator':
         """
         Utility method to access the id generator of the environment containe by this object
 

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

@@ -32,7 +32,7 @@
 import datetime
 import hashlib
 import codecs
-import pickle
+import pickle  # nosec: This is e controled pickle loading
 import typing
 import logging
 
@@ -61,7 +61,7 @@ class Cache:
         self._bowner = self._owner.encode('utf8')
 
     def __getKey(self, key: typing.Union[str, bytes]) -> str:
-        h = hashlib.md5()
+        h = hashlib.md5()  # nosec: not used for cryptography, just for hashing
         if isinstance(key, str):
             key = key.encode('utf8')
         h.update(self._bowner + key)
@@ -82,7 +82,7 @@ class Cache:
 
             try:
                 # logger.debug('value: %s', c.value)
-                val = pickle.loads(
+                val = pickle.loads(  # nosec: This is e controled pickle loading
                     typing.cast(bytes, codecs.decode(c.value.encode(), 'base64'))
                 )
             except Exception:  # If invalid, simple do no tuse it
@@ -127,7 +127,7 @@ class Cache:
         """
         self.remove(key)
 
-    def clean(self) -> None:
+    def clear(self) -> None:
         Cache.delete(self._owner)
 
     def put(

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

@@ -29,7 +29,7 @@
 """
 @author: Adolfo Gómez, dkmaster at dkmon dot com
 """
-import pickle
+import pickle    # nosec: This is e controled pickle use
 import base64
 import hashlib
 import codecs
@@ -46,7 +46,7 @@ MARK = '_mgb_'
 
 
 def _calcKey(owner: bytes, key: bytes, extra: typing.Optional[bytes] = None) -> str:
-    h = hashlib.md5()
+    h = hashlib.md5()  # nosec: not used for cryptography, just for hashing
     h.update(owner)
     h.update(key)
     if extra:
@@ -66,7 +66,7 @@ def _decodeValue(
 ) -> typing.Tuple[str, typing.Any]:
     if value:
         try:
-            v = pickle.loads(base64.b64decode(value.encode()))
+            v = pickle.loads(base64.b64decode(value.encode()))  # nosec: This is e controled pickle loading
             if isinstance(v, tuple) and v[0] == MARK:
                 return typing.cast(typing.Tuple[str, typing.Any], v[1:])
             # Fix value so it contains also the "key" (in this case, the original key is lost, we have only the hash value...)
@@ -312,7 +312,7 @@ class Storage:
     def getPickle(self, skey: typing.Union[str, bytes]) -> typing.Any:
         v = self.readData(skey, True)
         if v:
-            return pickle.loads(typing.cast(bytes, v))
+            return pickle.loads(typing.cast(bytes, v))  # nosec: This is e controled pickle loading
         return None
 
     def getPickleByAttr1(self, attr1: str, forUpdate: bool = False):
@@ -320,7 +320,7 @@ class Storage:
             query = DBStorage.objects.filter(owner=self._owner, attr1=attr1)
             if forUpdate:
                 query = query.select_for_update()
-            return pickle.loads(
+            return pickle.loads(  # nosec: This is e controled pickle loading
                 codecs.decode(query[0].data.encode(), 'base64')
             )  # @UndefinedVariable
         except Exception:
@@ -335,7 +335,7 @@ class Storage:
         try:
             # Process several keys at once
             DBStorage.objects.filter(key__in=[self.getKey(k) for k in keys]).delete()
-        except Exception:
+        except Exception:  # nosec: Not interested in processing exceptions, just ignores it
             pass
 
     def lock(self):
@@ -393,10 +393,10 @@ class Storage:
         self, attr1: typing.Optional[str] = None, forUpdate: bool = False
     ) -> typing.Iterable[typing.Tuple[str, typing.Any, str]]:
         for v in self.filter(attr1, forUpdate):
-            yield (v[0], pickle.loads(v[1]), v[2])
+            yield (v[0], pickle.loads(v[1]), v[2])  # nosec: secure pickle load
 
-    def clean(self):
+    def clear(self):
-        self.delete(self._owner)
+        Storage.delete(self._owner)
 
     @staticmethod
     def delete(owner: str) -> None:

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

@@ -69,7 +69,7 @@ class CacheTests(TestCase):
 
         # Checks cache clean
         cache.put('key', VALUE_1)
-        cache.clean()
+        cache.clear()
         self.assertEqual(cache.get('key'), None, 'Get key from cleaned cache')
 
         # Checks cache purge