Merge pull request greenbone#462 from greenbone/dependabot/pip/pylint…

…-2.11.1

Bump pylint from 2.10.2 to 2.11.1

ospd/command/command.py

@@ -85,8 +85,9 @@ def as_dict(self):
         }
 
     def __repr__(self):
-        return '<{} description="{}" attributes={} elements={}>'.format(
-            self.name, self.description, self.attributes, self.elements
+        return (
+            f'<{self.name} description="{self.description}" '
+            f'attributes={self.attributes} elements={self.elements}>'
         )
 
 
@@ -235,7 +236,7 @@ def handle_xml(self, xml: Element) -> bytes:
             output = subprocess.check_output(cmd)
         except (subprocess.CalledProcessError, OSError) as e:
             raise OspdCommandError(
-                'Bogus get_performance format. %s' % e, 'get_performance'
+                f'Bogus get_performance format. {e}', 'get_performance'
             ) from None
 
         return simple_response_str(
@@ -292,7 +293,7 @@ def handle_xml(self, xml: Element) -> bytes:
             )
 
         if not self._daemon.scan_exists(scan_id):
-            text = "Failed to find scan '{0}'".format(scan_id)
+            text = f"Failed to find scan '{scan_id}'"
             return simple_response_str('delete_scan', 404, text)
 
         self._daemon.check_scan_process(scan_id)
@@ -336,7 +337,7 @@ def handle_xml(self, xml: Element) -> Iterator[bytes]:
 
         if self._daemon.vts and vt_id and vt_id not in self._daemon.vts:
             self._daemon.vts.is_cache_available = True
-            text = "Failed to find vulnerability test '{0}'".format(vt_id)
+            text = f"Failed to find vulnerability test '{vt_id}'"
             raise OspdCommandError(text, 'get_vts', 404)
 
         filtered_vts = None
@@ -455,7 +456,7 @@ def handle_xml(self, xml: Element) -> bytes:
             )
             responses.append(scan)
         else:
-            text = "Failed to find scan '{0}'".format(scan_id)
+            text = f"Failed to find scan '{scan_id}'"
             return simple_response_str('get_scans', 404, text)
 
         return simple_response_str('get_scans', 200, 'OK', responses)
@@ -507,8 +508,8 @@ def handle_xml(self, xml: Element) -> bytes:
                 self._daemon.max_queued_scans,
             )
             raise OspdCommandError(
-                'Maximum number of queued scans set to %d reached.'
-                % self._daemon.max_queued_scans,
+                'Maximum number of queued scans set to '
+                f'{str(self._daemon.max_queued_scans)} reached.',
                 'start_scan',
             )
 

ospd/datapickler.py

@@ -69,15 +69,15 @@ def store_data(self, filename: str, data_object: Any) -> str:
             parent_dir.mkdir(parents=True, exist_ok=True)
         except Exception as e:
             raise OspdCommandError(
-                'Not possible to access dir for %s. %s' % (filename, e),
+                f'Not possible to access dir for {filename}. {e}',
                 'start_scan',
             ) from e
 
         try:
             pickled_data = pickle.dumps(data_object)
         except pickle.PicklingError as e:
             raise OspdCommandError(
-                'Not possible to pickle scan info for %s. %s' % (filename, e),
+                f'Not possible to pickle scan info for {filename}. {e}',
                 'start_scan',
             ) from e
 
@@ -89,7 +89,7 @@ def store_data(self, filename: str, data_object: Any) -> str:
         except Exception as e:  # pylint: disable=broad-except
             self._fd_close()
             raise OspdCommandError(
-                'Not possible to store scan info for %s. %s' % (filename, e),
+                f'Not possible to store scan info for {filename}. {e}',
                 'start_scan',
             ) from e
         self._fd_close()

ospd/errors.py

@@ -37,9 +37,7 @@ def __init__(self, function: str, argument: str) -> None:
         self.argument = argument
 
     def __str__(self) -> str:
-        return "{}: Argument {} is required".format(
-            self.function, self.argument
-        )
+        return f"{self.function}: Argument {self.argument} is required"
 
 
 class OspdCommandError(OspdError):

ospd/main.py

@@ -39,7 +39,7 @@
 This is free software: you are free to change and redistribute it.
 There is NO WARRANTY, to the extent permitted by law."""
 
-LOGGER = logging.getLogger(__name__)
+logger = logging.getLogger(__name__)
 
 
 def print_version(daemon: OSPDaemon, file=sys.stdout):
@@ -52,11 +52,11 @@ def print_version(daemon: OSPDaemon, file=sys.stdout):
     daemon_version = daemon.get_daemon_version()
 
     print(
-        "OSP Server for {0}: {1}".format(scanner_name, server_version),
+        f"OSP Server for {scanner_name}: {server_version}",
         file=file,
     )
-    print("OSP: {0}".format(protocol_version), file=file)
-    print("{0}: {1}".format(daemon_name, daemon_version), file=file)
+    print(f"OSP: {protocol_version}", file=file)
+    print(f"{daemon_name}: {daemon_version}", file=file)
     print(file=file)
     print(COPYRIGHT, file=file)
 
@@ -75,13 +75,13 @@ def exit_cleanup(
     if not pidpath.is_file():
         return
 
-    with pidpath.open() as f:
+    with pidpath.open(encoding='utf-8') as f:
         if int(f.read()) == os.getpid():
-            LOGGER.debug("Performing exit clean up")
+            logger.debug("Performing exit clean up")
             daemon.daemon_exit_cleanup()
-            LOGGER.info("Shutting-down server ...")
+            logger.info("Shutting-down server ...")
             server.close()
-            LOGGER.debug("Finishing daemon process")
+            logger.debug("Finishing daemon process")
             pidpath.unlink()
             sys.exit()
 
@@ -152,7 +152,7 @@ def main(
     if not daemon.check():
         return 1
 
-    LOGGER.info(
+    logger.info(
         "Starting %s version %s.",
         daemon.daemon_info['name'],
         daemon.daemon_info['version'],

ospd/misc.py

@@ -31,7 +31,7 @@
 
 import psutil
 
-LOGGER = logging.getLogger(__name__)
+logger = logging.getLogger(__name__)
 
 
 def create_process(
@@ -61,7 +61,7 @@ def get_str(cls, result_type: int) -> str:
         elif result_type == cls.HOST_DETAIL:
             return "Host Detail"
         else:
-            assert False, "Erroneous result type {0}.".format(result_type)
+            assert False, f"Erroneous result type {result_type}."
 
     @classmethod
     def get_type(cls, result_name: str) -> int:
@@ -75,7 +75,7 @@ def get_type(cls, result_name: str) -> int:
         elif result_name == "Host Detail":
             return cls.HOST_DETAIL
         else:
-            assert False, "Erroneous result name {0}.".format(result_name)
+            assert False, f"Erroneous result name {result_name}."
 
 
 def valid_uuid(value) -> bool:
@@ -94,7 +94,7 @@ def go_to_background() -> None:
         if os.fork():
             sys.exit()
     except OSError as errmsg:
-        LOGGER.error('Fork failed: %s', errmsg)
+        logger.error('Fork failed: %s', errmsg)
         sys.exit(1)
 
 
@@ -110,7 +110,7 @@ def create_pid(pidfile: str) -> bool:
 
     if pidpath.is_file():
         process_name = None
-        with pidpath.open('r') as pidfile:
+        with pidpath.open('r', encoding='utf-8') as pidfile:
             current_pid = pidfile.read()
             try:
                 process = psutil.Process(int(current_pid))
@@ -119,13 +119,13 @@ def create_pid(pidfile: str) -> bool:
                 pass
 
         if process_name == new_process_name:
-            LOGGER.error(
+            logger.error(
                 "There is an already running process. See %s.",
                 str(pidpath.absolute()),
             )
             return False
         else:
-            LOGGER.debug(
+            logger.debug(
                 "There is an existing pid file '%s', but the PID %s belongs to "
                 "the process %s. It seems that %s was abruptly stopped. "
                 "Removing the pid file.",
@@ -136,10 +136,10 @@ def create_pid(pidfile: str) -> bool:
             )
 
     try:
-        with pidpath.open(mode='w') as f:
+        with pidpath.open(mode='w', encoding='utf-8') as f:
             f.write(pid)
     except (FileNotFoundError, PermissionError) as e:
-        LOGGER.error(
+        logger.error(
             "Failed to create pid file %s. %s", str(pidpath.absolute()), e
         )
         return False

ospd/network.py

@@ -28,7 +28,7 @@
 
 from typing import List, Optional, Tuple
 
-__LOGGER = logging.getLogger(__name__)
+logger = logging.getLogger(__name__)
 
 
 def target_to_ipv4(target: str) -> Optional[List]:
@@ -303,7 +303,7 @@ def target_str_to_list(target_str: str) -> Optional[List]:
         if target_list:
             new_list.extend(target_list)
         else:
-            __LOGGER.info("%s: Invalid target value", target)
+            logger.info("%s: Invalid target value", target)
             return []
 
     return list(collections.OrderedDict.fromkeys(new_list))
@@ -370,7 +370,7 @@ def port_range_expand(portrange: str) -> Optional[List]:
         port_range_min = int(portrange[: portrange.index('-')])
         port_range_max = int(portrange[portrange.index('-') + 1 :]) + 1
     except (IndexError, ValueError) as e:
-        __LOGGER.info("Invalid port range format %s", e)
+        logger.info("Invalid port range format %s", e)
         return None
 
     port_list = list()
@@ -413,7 +413,7 @@ def ports_str_check_failed(port_str: str) -> bool:
         or port_str[len(port_str) - 1] == '-'
         or port_str.count(':') < (port_str.count('T') + port_str.count('U'))
     ):
-        __LOGGER.error("Invalid port range format")
+        logger.error("Invalid port range format")
         return True
 
     index = 0
@@ -423,7 +423,7 @@ def ports_str_check_failed(port_str: str) -> bool:
                 int(port_str[index - 1])
                 int(port_str[index + 1])
             except (TypeError, ValueError) as e:
-                __LOGGER.error("Invalid port range format: %s", e)
+                logger.error("Invalid port range format: %s", e)
                 return True
         index += 1
 
@@ -440,11 +440,11 @@ def ports_as_list(port_str: str) -> Tuple[Optional[List], Optional[List]]:
     @return two list of sorted integers, for tcp and udp ports respectively.
     """
     if not port_str:
-        __LOGGER.info("Invalid port value")
+        logger.info("Invalid port value")
         return [None, None]
 
     if ports_str_check_failed(port_str):
-        __LOGGER.info("{0}: Port list malformed.")
+        logger.info("{0}: Port list malformed.")
         return [None, None]
 
     tcp_list = list()
@@ -522,7 +522,7 @@ def port_list_compress(port_list: List) -> str:
     """Compress a port list and return a string."""
 
     if not port_list or len(port_list) == 0:
-        __LOGGER.info("Invalid or empty port list.")
+        logger.info("Invalid or empty port list.")
         return ''
 
     port_list = sorted(set(port_list))

ospd/ospd.py

@@ -305,23 +305,19 @@ def preprocess_scan_params(self, xml_params):
                     params[key] = int(params[key])
                 except ValueError:
                     raise OspdCommandError(
-                        'Invalid %s value' % key, 'start_scan'
+                        f'Invalid {key} value', 'start_scan'
                     ) from None
 
             if param_type == 'boolean':
                 if params[key] not in [0, 1]:
-                    raise OspdCommandError(
-                        'Invalid %s value' % key, 'start_scan'
-                    )
+                    raise OspdCommandError(f'Invalid {key} value', 'start_scan')
             elif param_type == 'selection':
                 selection = self.get_scanner_param_default(key).split('|')
                 if params[key] not in selection:
-                    raise OspdCommandError(
-                        'Invalid %s value' % key, 'start_scan'
-                    )
+                    raise OspdCommandError(f'Invalid {key} value', 'start_scan')
             if self.get_scanner_param_mandatory(key) and params[key] == '':
                 raise OspdCommandError(
-                    'Mandatory %s value is missing' % key, 'start_scan'
+                    f'Mandatory {key} value is missing', 'start_scan'
                 )
 
         return params
@@ -343,9 +339,7 @@ def stop_scan(self, scan_id: str) -> None:
 
         scan_process = self.scan_processes.get(scan_id)
         if not scan_process:
-            raise OspdCommandError(
-                'Scan not found {0}.'.format(scan_id), 'stop_scan'
-            )
+            raise OspdCommandError(f'Scan not found {scan_id}.', 'stop_scan')
         if not scan_process.is_alive():
             raise OspdCommandError(
                 'Scan already stopped or finished.', 'stop_scan'
@@ -532,7 +526,7 @@ def start_scan(self, scan_id: str) -> None:
                 scan_id,
                 name='',
                 host=self.get_scan_host(scan_id),
-                value='Host process failure (%s).' % e,
+                value=f'Host process failure ({e}).',
             )
             logger.exception('%s: Exception %s while scanning', scan_id, e)
         else:
@@ -562,7 +556,7 @@ def handle_timeout(self, scan_id: str, host: str) -> None:
             scan_id,
             host=host,
             name="Timeout",
-            value="{0} exec timeout.".format(self.get_scanner_name()),
+            value=f"{self.get_scanner_name()} exec timeout.",
         )
 
     def sort_host_finished(
@@ -681,13 +675,13 @@ def get_help_text(self) -> str:
             attributes = info.get_attributes()
             elements = info.get_elements()
 
-            command_txt = "\t{0: <22} {1}\n".format(name, description)
+            command_txt = f"\t{name: <22} {description}\n"
 
             if attributes:
                 command_txt = ''.join([command_txt, "\t Attributes:\n"])
 
                 for attrname, attrdesc in attributes.items():
-                    attr_txt = "\t  {0: <22} {1}\n".format(attrname, attrdesc)
+                    attr_txt = f"\t  {attrname: <22} {attrdesc}\n"
                     command_txt = ''.join([command_txt, attr_txt])
 
             if elements:
@@ -1211,7 +1205,7 @@ def handle_command(self, data: bytes, stream: Stream) -> None:
 
         if not self.initialized and command.must_be_initialized:
             exception = OspdCommandError(
-                '%s is still starting' % self.daemon_info['name'], 'error'
+                f'{self.daemon_info["name"]} is still starting', 'error'
             )
             response = exception.as_xml()
             stream.write(response)

ospd/ospd_ssh.py

@@ -88,7 +88,7 @@ def __init__(self, **kwargs):
         if paramiko is None:
             raise ImportError(
                 'paramiko needs to be installed in order to use'
-                ' the %s class.' % self.__class__.__name__
+                f' the {self.__class__.__name__} class.'
             )
 
         for name, param in SSH_SCANNER_PARAMS.items():
@@ -154,7 +154,7 @@ def run_command(self, scan_id: str, host: str, cmd: str) -> Optional[str]:
             return None
 
         if self._niceness is not None:
-            cmd = "nice -n %s %s" % (self._niceness, cmd)
+            cmd = f"nice -n {self._niceness} {cmd}"
         _, stdout, _ = ssh.exec_command(cmd)
         result = stdout.readlines()
         ssh.close()