Add nosec pragma for bandit false positives. Also re-format some of the

code to comply with style rules.

Author: Kami

libcloud/common/gandi.py

@@ -36,10 +36,12 @@ class GandiException(Exception):
 
     def __str__(self):
         # pylint: disable=unsubscriptable-object
+
         return "({}) {}".format(self.args[0], self.args[1])
 
     def __repr__(self):
         # pylint: disable=unsubscriptable-object
+
         return '<GandiException code {} "{}">'.format(self.args[0], self.args[1])
 
 
@@ -83,6 +85,7 @@ def __init__(
 
     def request(self, method, *args):
         args = (self.key,) + args
+
         return super().request(method, *args)
 
 
@@ -106,6 +109,7 @@ def _wait_operation(self, id, timeout=DEFAULT_TIMEOUT, check_interval=DEFAULT_IN
 
                 if op["step"] == "DONE":
                     return True
+
                 if op["step"] in ["ERROR", "CANCEL"]:
                     return False
             except (KeyError, IndexError):
@@ -114,6 +118,7 @@ def _wait_operation(self, id, timeout=DEFAULT_TIMEOUT, check_interval=DEFAULT_IN
                 raise GandiException(1002, e)
 
             time.sleep(check_interval)
+
         return False
 
 
@@ -147,7 +152,8 @@ def get_uuid(self):
         same UUID!
         """
         hashstring = "{}:{}:{}".format(self.uuid_prefix, self.id, self.driver.type)
-        return hashlib.sha1(b(hashstring)).hexdigest()
+
+        return hashlib.sha1(b(hashstring)).hexdigest()  # nosec
 
 
 class IPAddress(BaseObject):

libcloud/common/nfsn.py

@@ -45,10 +45,13 @@ def parse_error(self):
         # If we only have one of "error" or "debug", use the one that we have.
         # If we have both, use both, with a space character in between them.
         value = "No message specified"
+
         if error is not None:
             value = error
+
         if debug is not None:
             value = debug
+
         if error is not None and value is not None:
             value = error + " " + value
         value = value + " (HTTP Code: %d)" % self.status
@@ -70,22 +73,25 @@ def _header(self, action, data):
         salt = self._salt()
         api_key = self.key
         data = urlencode(data)
-        data_hash = hashlib.sha1(data.encode("utf-8")).hexdigest()
+        data_hash = hashlib.sha1(data.encode("utf-8")).hexdigest()  # nosec
 
         string = ";".join((login, timestamp, salt, api_key, action, data_hash))
-        string_hash = hashlib.sha1(string.encode("utf-8")).hexdigest()
+        string_hash = hashlib.sha1(string.encode("utf-8")).hexdigest()  # nosec
 
         return ";".join((login, timestamp, salt, string_hash))
 
     def request(self, action, params=None, data="", headers=None, method="GET"):
         """Add the X-NFSN-Authentication header to an HTTP request."""
+
         if not headers:
             headers = {}
+
         if not params:
             params = {}
         header = self._header(action, data)
 
         headers["X-NFSN-Authentication"] = header
+
         if method == "POST":
             headers["Content-Type"] = "application/x-www-form-urlencoded"
 
@@ -94,16 +100,20 @@ def request(self, action, params=None, data="", headers=None, method="GET"):
     def encode_data(self, data):
         """NFSN expects the body to be regular key-value pairs that are not
         JSON-encoded."""
+
         if data:
             data = urlencode(data)
+
         return data
 
     def _salt(self):
         """Return a 16-character alphanumeric string."""
         r = random.SystemRandom()
+
         return "".join(r.choice(SALT_CHARACTERS) for _ in range(16))
 
     def _timestamp(self):
         """Return the current number of seconds since the Unix epoch,
         as a string."""
+
         return str(int(time.time()))

libcloud/common/ovh.py

@@ -109,11 +109,13 @@ class OvhConnection(ConnectionUserAndKey):
 
     def __init__(self, user_id, *args, **kwargs):
         region = kwargs.pop("region", "")
+
         if region:
             self.host = ("{}.{}".format(region, API_HOST)).lstrip(".")
         else:
             self.host = API_HOST
         self.consumer_key = kwargs.pop("ex_consumer_key", None)
+
         if self.consumer_key is None:
             consumer_key_json = self.request_consumer_key(user_id)
             msg = (
@@ -145,27 +147,32 @@ def request_consumer_key(self, user_id):
 
         json_response = response.parse_body()
         httpcon.close()
+
         return json_response
 
     def get_timestamp(self):
         if not self._timedelta:
             url = "https://{}{}/auth/time".format(self.host, API_ROOT)
             response = get_response_object(url=url, method="GET", headers={})
+
             if not response or not response.body:
                 raise Exception("Failed to get current time from Ovh API")
 
             timestamp = int(response.body)
             self._timedelta = timestamp - int(time.time())
+
         return int(time.time()) + self._timedelta
 
     def make_signature(self, method, action, params, data, timestamp):
         full_url = "https://{}{}".format(self.host, action)
+
         if params:
             full_url += "?"
+
             for key, value in params.items():
                 full_url += "{}={}&".format(key, value)
             full_url = full_url[:-1]
-        sha1 = hashlib.sha1()
+        sha1 = hashlib.sha1()  # nosec
         base_signature = "+".join(
             [
                 self.key,
@@ -178,6 +185,7 @@ def make_signature(self, method, action, params, data, timestamp):
         )
         sha1.update(base_signature.encode())
         signature = "$1$" + sha1.hexdigest()
+
         return signature
 
     def add_default_params(self, params):
@@ -191,6 +199,7 @@ def add_default_headers(self, headers):
                 "Content-type": "application/json",
             }
         )
+
         return headers
 
     def request(self, action, params=None, data=None, headers=None, method="GET", raw=False):

libcloud/compute/base.py

@@ -144,14 +144,18 @@ def get_uuid(self):
 
         :rtype: ``str``
         """
+
         if not self._uuid:
-            self._uuid = hashlib.sha1(b("{}:{}".format(self.id, self.driver.type))).hexdigest()
+            self._uuid = hashlib.sha1(
+                b("{}:{}".format(self.id, self.driver.type))
+            ).hexdigest()  # nosec
 
         return self._uuid
 
     @property
     def uuid(self):
         # type: () -> str
+
         return self.get_uuid()
 
 
@@ -280,6 +284,7 @@ def reboot(self):
         >>> node.state == NodeState.REBOOTING
         True
         """
+
         return self.driver.reboot_node(self)
 
     def start(self):
@@ -289,6 +294,7 @@ def start(self):
 
         :return: ``bool``
         """
+
         return self.driver.start_node(self)
 
     def stop_node(self):
@@ -298,6 +304,7 @@ def stop_node(self):
 
         :return: ``bool``
         """
+
         return self.driver.stop_node(self)
 
     def destroy(self):
@@ -321,6 +328,7 @@ def destroy(self):
         False
 
         """
+
         return self.driver.destroy_node(self)
 
     def __repr__(self):
@@ -688,6 +696,7 @@ def list_snapshots(self):
         """
         :rtype: ``list`` of ``VolumeSnapshot``
         """
+
         return self.driver.list_volume_snapshots(volume=self)
 
     def attach(self, node, device=None):
@@ -705,6 +714,7 @@ def attach(self, node, device=None):
         :return: ``True`` if attach was successful, ``False`` otherwise.
         :rtype: ``bool``
         """
+
         return self.driver.attach_volume(node=node, volume=self, device=device)
 
     def detach(self):
@@ -715,6 +725,7 @@ def detach(self):
         :return: ``True`` if detach was successful, ``False`` otherwise.
         :rtype: ``bool``
         """
+
         return self.driver.detach_volume(volume=self)
 
     def snapshot(self, name):
@@ -725,6 +736,7 @@ def snapshot(self, name):
         :return: Created snapshot.
         :rtype: ``VolumeSnapshot``
         """
+
         return self.driver.create_volume_snapshot(volume=self, name=name)
 
     def destroy(self):
@@ -804,6 +816,7 @@ def destroy(self):
 
         :rtype: ``bool``
         """
+
         return self.driver.destroy_volume_snapshot(snapshot=self)
 
     def __repr__(self):
@@ -1148,6 +1161,7 @@ def deploy_node(
         :type wait_period: ``int``
 
         """
+
         if not libcloud.compute.ssh.have_paramiko:
             raise RuntimeError(
                 "paramiko is not installed. You can install " + "it using pip: pip install paramiko"
@@ -1163,6 +1177,7 @@ def deploy_node(
             pass
         elif "create_node" in self.features:
             f = self.features["create_node"]
+
             if "generates_password" not in f and "password" not in f:
                 raise NotImplementedError("deploy_node not implemented for this driver")
         else:
@@ -1181,13 +1196,15 @@ def deploy_node(
         try:
             # NOTE: We only pass auth to the method if auth argument is
             # provided
+
             if auth:
                 node = self.create_node(auth=auth, **create_node_kwargs)
             else:
                 node = self.create_node(**create_node_kwargs)
         except TypeError as e:
             msg_1_re = r"create_node\(\) missing \d+ required " "positional arguments.*"
             msg_2_re = r"create_node\(\) takes at least \d+ arguments.*"
+
             if re.match(msg_1_re, str(e)) or re.match(msg_2_re, str(e)):
                 # pylint: disable=unexpected-keyword-arg
                 node = self.create_node(  # type: ignore
@@ -1211,6 +1228,7 @@ def deploy_node(
             atexit.register(at_exit_func, driver=self, node=node)
 
         password = None
+
         if auth:
             if isinstance(auth, NodeAuthPassword):
                 password = auth.password
@@ -1261,6 +1279,7 @@ def deploy_node(
             else:
                 # Script successfully executed, don't try alternate username
                 deploy_error = None
+
                 break
 
         if deploy_error is not None:
@@ -1674,15 +1693,18 @@ def is_supported(address):
             """
             Return True for supported address.
             """
+
             if force_ipv4 and not is_valid_ip_address(address=address, family=socket.AF_INET):
                 return False
+
             return True
 
         def filter_addresses(addresses):
             # type: (List[str]) -> List[str]
             """
             Return list of supported addresses.
             """
+
             return [address for address in addresses if is_supported(address)]
 
         if ssh_interface not in ["public_ips", "private_ips"]:
@@ -1708,15 +1730,18 @@ def filter_addresses(addresses):
 
             running_nodes = [node for node in matching_nodes if node.state == NodeState.RUNNING]
             addresses = []
+
             for node in running_nodes:
                 node_addresses = filter_addresses(getattr(node, ssh_interface))
+
                 if len(node_addresses) >= 1:
                     addresses.append(node_addresses)
 
             if len(running_nodes) == len(uuids) == len(addresses):
                 return list(zip(running_nodes, addresses))
             else:
                 time.sleep(wait_period)
+
                 continue
 
         raise LibcloudError(value="Timed out after %s seconds" % (timeout), driver=self)
@@ -1757,6 +1782,7 @@ def _get_and_check_auth(self, auth):
             # Some providers require password to also include uppercase
             # characters so convert some characters to uppercase
             password = ""
+
             for char in value:
                 if not char.isdigit() and char.islower():
                     if random.randint(0, 1) == 1:
@@ -1784,6 +1810,7 @@ def _wait_until_running(
         # type: (Node, float, int, str, bool) -> List[Tuple[Node, List[str]]]
         # This is here for backward compatibility and will be removed in the
         # next major release
+
         return self.wait_until_running(
             nodes=[node],
             wait_period=wait_period,
@@ -1836,6 +1863,7 @@ def _ssh_client_connect(self, ssh_client, wait_period=1.5, timeout=300):
                     pass
 
                 time.sleep(wait_period)
+
                 continue
             else:
                 return ssh_client
@@ -1881,6 +1909,7 @@ def _connect_and_run_deployment_script(
         node = self._run_deployment_script(
             task=task, node=node, ssh_client=ssh_client, max_tries=max_tries
         )
+
         return node
 
     def _run_deployment_script(self, task, node, ssh_client, max_tries=3):
@@ -1940,6 +1969,7 @@ def _run_deployment_script(self, task, node, ssh_client, max_tries=3):
             else:
                 # Deployment succeeded
                 ssh_client.close()
+
                 return node
 
         return node
@@ -1949,6 +1979,7 @@ def _get_size_price(self, size_id):
         """
         Return pricing information for the provided size id.
         """
+
         return get_size_price(driver_type="compute", driver_name=self.api_name, size_id=size_id)