bandit security warning fixes.

contrib/packs/tests/test_action_download.py

@@ -20,6 +20,10 @@
 import shutil
 import tempfile
 import hashlib
+import sys
+
+# TODO: Move keywords directly to hashlib.md5 call as part of dropping py3.8.
+hashlib_kwargs = {} if sys.version_info[0:2] < (3, 9) else {"usedforsecurity": False}
 
 from st2common.util.monkey_patch import use_select_poll_workaround
 
@@ -154,7 +158,9 @@ def tearDown(self):
     def test_run_pack_download(self):
         action = self.get_action_instance()
         result = action.run(packs=["test"], abs_repo_base=self.repo_base)
-        temp_dir = hashlib.md5(PACK_INDEX["test"]["repo_url"].encode()).hexdigest()
+        temp_dir = hashlib.md5(
+            PACK_INDEX["test"]["repo_url"].encode(), **hashlib_kwargs
+        ).hexdigest()  # nosec. remove nosec after py3.8 drop
 
         self.assertEqual(result, {"test": "Success."})
         self.clone_from.assert_called_once_with(
@@ -175,8 +181,12 @@ def test_run_pack_download_dependencies(self):
             abs_repo_base=self.repo_base,
         )
         temp_dirs = [
-            hashlib.md5(PACK_INDEX["test2"]["repo_url"].encode()).hexdigest(),
-            hashlib.md5(PACK_INDEX["test4"]["repo_url"].encode()).hexdigest(),
+            hashlib.md5(
+                PACK_INDEX["test2"]["repo_url"].encode(), **hashlib_kwargs
+            ).hexdigest(),  # nosec. remove nosec after py3.8 drop
+            hashlib.md5(
+                PACK_INDEX["test4"]["repo_url"].encode(), **hashlib_kwargs
+            ).hexdigest(),  # nosec. remove nosec after py3.8 drop
         ]
 
         self.assertEqual(result, {"test2": "Success.", "test4": "Success."})
@@ -205,8 +215,12 @@ def test_run_pack_download_multiple_packs(self):
         action = self.get_action_instance()
         result = action.run(packs=["test", "test2"], abs_repo_base=self.repo_base)
         temp_dirs = [
-            hashlib.md5(PACK_INDEX["test"]["repo_url"].encode()).hexdigest(),
-            hashlib.md5(PACK_INDEX["test2"]["repo_url"].encode()).hexdigest(),
+            hashlib.md5(
+                PACK_INDEX["test"]["repo_url"].encode(), **hashlib_kwargs
+            ).hexdigest(),  # nosec. remove nosec after py3.8 drop
+            hashlib.md5(
+                PACK_INDEX["test2"]["repo_url"].encode(), **hashlib_kwargs
+            ).hexdigest(),  # nosec. remove nosec after py3.8 drop
         ]
 
         self.assertEqual(result, {"test": "Success.", "test2": "Success."})
@@ -243,7 +257,9 @@ def test_run_pack_download_no_tag(self):
 
     def test_run_pack_lock_is_already_acquired(self):
         action = self.get_action_instance()
-        temp_dir = hashlib.md5(PACK_INDEX["test"]["repo_url"].encode()).hexdigest()
+        temp_dir = hashlib.md5(
+            PACK_INDEX["test"]["repo_url"].encode(), **hashlib_kwargs
+        ).hexdigest()  # nosec. remove nosec after py3.8 drop
 
         original_acquire = LockFile.acquire
 
@@ -274,7 +290,9 @@ def mock_acquire(self, timeout=None):
     def test_run_pack_lock_is_already_acquired_force_flag(self):
         # Lock is already acquired but force is true so it should be deleted and released
         action = self.get_action_instance()
-        temp_dir = hashlib.md5(PACK_INDEX["test"]["repo_url"].encode()).hexdigest()
+        temp_dir = hashlib.md5(
+            PACK_INDEX["test"]["repo_url"].encode(), **hashlib_kwargs
+        ).hexdigest()  # nosec. remove nosec after py3.8 drop
 
         original_acquire = LockFile.acquire
 
@@ -682,7 +700,9 @@ def test_run_pack_download_local_directory(self):
     def test_run_pack_download_with_tag(self):
         action = self.get_action_instance()
         result = action.run(packs=["test"], abs_repo_base=self.repo_base)
-        temp_dir = hashlib.md5(PACK_INDEX["test"]["repo_url"].encode()).hexdigest()
+        temp_dir = hashlib.md5(
+            PACK_INDEX["test"]["repo_url"].encode(), **hashlib_kwargs
+        ).hexdigest()  # nosec. remove nosec after py3.8 drop
 
         self.assertEqual(result, {"test": "Success."})
         self.clone_from.assert_called_once_with(

st2common/st2common/models/db/trace.py

@@ -15,6 +15,10 @@
 
 from __future__ import absolute_import
 import hashlib
+import sys
+
+# TODO: Move keywords directly to hashlib.md5 call as part of dropping py3.8.
+hashlib_kwargs = {} if sys.version_info[0:2] < (3, 9) else {"usedforsecurity": False}
 
 import mongoengine as me
 
@@ -107,7 +111,9 @@ def get_uid(self):
         parts = []
         parts.append(self.RESOURCE_TYPE)
 
-        components_hash = hashlib.md5()
+        components_hash = hashlib.md5(
+            **hashlib_kwargs
+        )  # nosec. remove nosec after py3.8 drop
         components_hash.update(str(self.trace_tag).encode())
         components_hash.update(str(self.trigger_instances).encode())
         components_hash.update(str(self.rules).encode())

st2common/st2common/models/db/trigger.py

@@ -17,6 +17,10 @@
 
 import json
 import hashlib
+import sys
+
+# TODO: Move keywords directly to hashlib.md5 call as part of dropping py3.8.
+hashlib_kwargs = {} if sys.version_info[0:2] < (3, 9) else {"usedforsecurity": False}
 
 import mongoengine as me
 
@@ -116,7 +120,9 @@ def get_uid(self):
         # compatibility reasons.
         parameters = getattr(self, "parameters", {})
         parameters = json.dumps(parameters, sort_keys=True)
-        parameters = hashlib.md5(parameters.encode()).hexdigest()
+        parameters = hashlib.md5(
+            parameters.encode(), **hashlib_kwargs
+        ).hexdigest()  # nosec. remove nosec after py3.8 drop
 
         uid = uid + self.UID_SEPARATOR + parameters
         return uid

st2common/st2common/util/crypto.py

@@ -39,6 +39,10 @@
 import base64
 
 from hashlib import sha1
+import sys
+
+# TODO: Move keywords directly to sha1 call as part of dropping py3.8.
+hashlib_kwargs = {} if sys.version_info[0:2] < (3, 9) else {"usedforsecurity": False}
 
 import six
 
@@ -70,7 +74,13 @@
 # Keyczar related constants
 KEYCZAR_HEADER_SIZE = 5
 KEYCZAR_AES_BLOCK_SIZE = 16
-KEYCZAR_HLEN = sha1().digest_size
+# usedforsecurity: False used here because KEYCZAR is deprecated
+# inherently insecure and will need to be removed from the code base when
+# the cryptography implementation is revised.  This is just to keep
+# bandit happy.
+KEYCZAR_HLEN = sha1(
+    **hashlib_kwargs
+).digest_size  # nosec. remove nosec after py3.8 drop
 
 # Minimum key size which can be used for symmetric crypto
 MINIMUM_AES_KEY_SIZE = 128

st2common/st2common/util/pack_management.py

@@ -25,6 +25,10 @@
 import hashlib
 import stat
 import re
+import sys
+
+# TODO: Move keywords directly to hashlib.md5 call as part of dropping py3.8.
+hashlib_kwargs = {} if sys.version_info[0:2] < (3, 9) else {"usedforsecurity": False}
 
 # This test workaround needs to be used before importing git
 from st2common.util.monkey_patch import use_select_poll_workaround
@@ -113,7 +117,9 @@ def download_pack(
 
     result = [pack_url, None, None]
 
-    temp_dir_name = hashlib.md5(pack_url.encode()).hexdigest()
+    temp_dir_name = hashlib.md5(
+        pack_url.encode(), **hashlib_kwargs
+    ).hexdigest()  # nosec. remove nosec after py3.8 drop
     lock_file = LockFile("/tmp/%s" % (temp_dir_name))
     lock_file_path = lock_file.lock_file
 

st2common/tests/unit/test_db_model_uids.py

@@ -17,6 +17,10 @@
 import json
 import hashlib
 from collections import OrderedDict
+import sys
+
+# TODO: Move keywords directly to hashlib.md5 call as part of dropping py3.8.
+hashlib_kwargs = {} if sys.version_info[0:2] < (3, 9) else {"usedforsecurity": False}
 
 import unittest
 
@@ -61,7 +65,9 @@ def test_get_uid(self):
         # Verify that same set of parameters always results in the same hash
         parameters = {"a": 1, "b": "unicode", "c": [1, 2, 3], "d": {"g": 1, "h": 2}}
         paramers_hash = json.dumps(parameters, sort_keys=True)
-        paramers_hash = hashlib.md5(paramers_hash.encode()).hexdigest()
+        paramers_hash = hashlib.md5(
+            paramers_hash.encode(), **hashlib_kwargs
+        ).hexdigest()  # nosec. remove nosec after py3.8 drop
 
         parameters = {"a": 1, "b": "unicode", "c": [1, 2, 3], "d": {"g": 1, "h": 2}}
         trigger_db = TriggerDB(name="tname", pack="tpack", parameters=parameters)

st2reactor/st2reactor/container/hash_partitioner.py

@@ -16,6 +16,10 @@
 from __future__ import absolute_import
 import ctypes
 import hashlib
+import sys
+
+# TODO: Move keywords directly to hashlib.md5 call as part of dropping py3.8.
+hashlib_kwargs = {} if sys.version_info[0:2] < (3, 9) else {"usedforsecurity": False}
 
 from st2reactor.container.partitioners import (
     DefaultPartitioner,
@@ -107,8 +111,10 @@ def _hash_sensor_ref(self, sensor_ref):
 
         # From http://www.cs.hmc.edu/~geoff/classes/hmc.cs070.200101/homework10/hashfuncs.html
         # The 'liberal' use of ctypes.c_unit is to guarantee unsigned integer and workaround
-        # inifinite precision.
-        md5_hash = hashlib.md5(sensor_ref.encode())
+        # infinite precision.
+        md5_hash = hashlib.md5(
+            sensor_ref.encode(), **hashlib_kwargs
+        )  # nosec. remove nosec after py3.8 drop
         md5_hash_int_repr = int(md5_hash.hexdigest(), 16)
         h = ctypes.c_uint(0)
         for d in reversed(str(md5_hash_int_repr)):