ENH: upgrade Python syntax to 3.6+

CHANGELOG.md

@@ -1,5 +1,5 @@
 <!-- comment
-   SPDX-FileCopyrightText: 2015-2021 Sebastian Wagner
+   SPDX-FileCopyrightText: 2015-2022 Sebastian Wagner
    SPDX-License-Identifier: AGPL-3.0-or-later
 -->
 
@@ -9,6 +9,8 @@ CHANGELOG
 3.1.0 (unreleased)
 ------------------
 
+- Upgraded syntax to Python 3.6 (mostly Format-Strings) using pyuprade (PR#2136 by Sebastian Wagner).
+
 ### Configuration
 
 ### Core

contrib/check_mk/cronjob_intelmq_statistics.py

@@ -31,6 +31,6 @@
             value = '0'
         else:
             value = value.decode()
-        stats.append("%s=%s" % (key.decode(), value))
+        stats.append(f"{key.decode()}={value}")
     handle.write("|".join(stats))
     handle.write('\n')

contrib/eventdb/apply_domain_suffix.py

@@ -20,7 +20,7 @@ def eventdb_apply(host, port,
                   table, dry_run, where,
                   filename):
     if password:
-        password = input('Password for user %r on %r: ' % (username, host))
+        password = input(f'Password for user {username!r} on {host!r}: ')
     where = 'AND ' + where if where else ''
 
     con1 = psycopg2.connect(user=username,

contrib/eventdb/apply_mapping_eventdb.py

@@ -50,7 +50,7 @@ def eventdb_apply(malware_name_column, malware_family_column, host, port,
         print("Error: Python module 'psycopg2' is needed but not available.", file=sys.stderr)
         return 2
     if password:
-        password = input('Password for user %r on %r: ' % (username, host))
+        password = input(f'Password for user {username!r} on {host!r}: ')
     where = 'AND ' + where if where else ''
 
     db = psycopg2.connect(database=database, user=username, password=password,

contrib/malware_name_mapping/download_mapping.py

@@ -38,7 +38,7 @@ def none_len(arg: Optional[list]):
 
 
 def generate_rule(expression: str, identifier: str, name=None):
-    return {"rulename": name if name else "%s-%s" % (identifier,
+    return {"rulename": name if name else "{}-{}".format(identifier,
                                                      hashlib.sha1(expression.encode()).hexdigest()[:10]),
             "if": {"classification.taxonomy": "malicious-code",
                    "malware.name": expression
@@ -112,7 +112,7 @@ def download(url: str = URL, add_default=False, params=None, include_malpedia=Fa
         rules.append(generate_rule(".*", add_default, name="default"))
 
     if params:
-        rules.extend((generate_rule(rule[0][0], rule[1][0]) for rule in params))
+        rules.extend(generate_rule(rule[0][0], rule[1][0]) for rule in params)
 
     return json.dumps(rules, indent=4, separators=(',', ': '))
 

docs/autogen.py

@@ -71,7 +71,7 @@ def harm_docs():
 
 
 def info(key, value=""):
-    return ("* **%s:** %s\n" % (key.title(), value)).strip() + '\n'
+    return (f"* **{key.title()}:** {value}\n").strip() + '\n'
 
 
 def feeds_docs():
@@ -134,7 +134,7 @@ def feeds_docs():
                         if isinstance(value, (list, tuple)) and value:
                             value = json.dumps(value)
 
-                        output += "   * `%s`: `%s`\n" % (key, value)
+                        output += f"   * `{key}`: `{value}`\n"
 
                 output += '\n'
 

intelmq/bin/intelmq_generate_misp_objects_templates.py

@@ -119,11 +119,11 @@ def dump_templates(self):
 
     objects = Path(args.objects)
     if not objects.exists():
-        raise Exception('Path to misp-objects repository does not exists: {args.objects}'.format(args=args))
+        raise Exception(f'Path to misp-objects repository does not exists: {args.objects}')
 
     harmonization_file = Path(args.harmonization)
     if not harmonization_file.exists():
-        raise Exception('Path to harmonization configuration does not exists: {args.harmonization}'.format(args=args))
+        raise Exception(f'Path to harmonization configuration does not exists: {args.harmonization}')
 
     g = MISPObjectTemplateGenerator(objects, harmonization_file)
     g.generate_templates()

intelmq/bin/intelmq_psql_initdb.py

@@ -28,9 +28,9 @@ def generate(harmonization_file=HARMONIZATION_CONF_FILE):
 
     try:
         print("INFO - Reading %s file" % harmonization_file)
-        with open(harmonization_file, 'r') as fp:
+        with open(harmonization_file) as fp:
             DATA = json.load(fp)['event']
-    except IOError:
+    except OSError:
         print("ERROR - Could not find %s" % harmonization_file)
         print("ERROR - Make sure that you have intelmq installed.")
         sys.exit(2)
@@ -69,7 +69,7 @@ def generate(harmonization_file=HARMONIZATION_CONF_FILE):
     initdb = """CREATE TABLE events (
     "id" BIGSERIAL UNIQUE PRIMARY KEY,"""
     for field, field_type in sorted(FIELDS.items()):
-        initdb += '\n    "{name}" {type},'.format(name=field, type=field_type)
+        initdb += f'\n    "{field}" {field_type},'
 
     initdb = initdb[:-1]  # remove last ','
     initdb += "\n);\n"
@@ -84,7 +84,7 @@ def main():
     fp = None
     try:
         if os.path.exists(OUTPUTFILE):
-            print('INFO - File {} exists, generating temporary file.'.format(OUTPUTFILE))
+            print(f'INFO - File {OUTPUTFILE} exists, generating temporary file.')
             os_fp, OUTPUTFILE = tempfile.mkstemp(suffix='.initdb.sql',
                                                  text=True)
             fp = os.fdopen(os_fp, 'wt')

intelmq/bin/intelmqctl.py

@@ -39,7 +39,7 @@
     psutil = None
 
 
-class Parameters(object):
+class Parameters:
     pass
 
 
@@ -206,7 +206,7 @@ def __init__(self, interactive: bool = False, returntype: ReturnType = ReturnTyp
         try:
             self._runtime_configuration = utils.load_configuration(RUNTIME_CONF_FILE)
         except ValueError as exc:  # pragma: no cover
-            self.abort('Error loading %r: %s' % (RUNTIME_CONF_FILE, exc))
+            self.abort(f'Error loading {RUNTIME_CONF_FILE!r}: {exc}')
 
         self._processmanagertype = getattr(self._parameters, 'process_manager', 'intelmq')
         if self._processmanagertype not in process_managers():
@@ -828,13 +828,13 @@ def check(self, no_connections=False):
         try:
             with open(HARMONIZATION_CONF_FILE) as file_handle:
                 files[HARMONIZATION_CONF_FILE] = json.load(file_handle)
-        except (IOError, ValueError) as exc:  # pragma: no cover
+        except (OSError, ValueError) as exc:  # pragma: no cover
             check_logger.error('Could not load %r: %s.', HARMONIZATION_CONF_FILE, exc)
             retval = 1
         try:
             with open(RUNTIME_CONF_FILE) as file_handle:
                 files[RUNTIME_CONF_FILE] = yaml.load(file_handle)
-        except (IOError, ValueError) as exc:
+        except (OSError, ValueError) as exc:
             check_logger.error('Could not load %r: %s.', RUNTIME_CONF_FILE, exc)
             retval = 1
         if retval:
@@ -933,7 +933,7 @@ def check(self, no_connections=False):
                 bot_check = bot.check(bot_parameters)
                 if bot_check:
                     for log_line in bot_check:
-                        getattr(check_logger, log_line[0])("Bot %r: %s" % (bot_id, log_line[1]))
+                        getattr(check_logger, log_line[0])(f"Bot {bot_id!r}: {log_line[1]}")
         for group in utils.list_all_bots().values():
             for bot_id, bot in group.items():
                 if subprocess.call(['which', bot['module']], stdout=subprocess.DEVNULL,
@@ -1026,7 +1026,7 @@ def upgrade_conf(self, previous=None, dry_run=None, function=None,
                 utils.write_configuration(state_file, state, new=True, useyaml=False)
             except Exception as exc:
                 self._logger.error('Error writing state file %r: %s.', state_file, exc)
-                return 1, 'Error writing state file %r: %s.' % (state_file, exc)
+                return 1, f'Error writing state file {state_file!r}: {exc}.'
             self._logger.info('Successfully wrote initial state file.')
 
         runtime = utils.load_configuration(RUNTIME_CONF_FILE)
@@ -1241,7 +1241,7 @@ def debug(self, sections=None):
                          'CONFIG_DIR', 'ROOT_DIR'):
                 output['paths'][path] = variables[path]
                 if self._returntype is ReturnType.TEXT:
-                    print('%s: %r' % (path, variables[path]))
+                    print(f'{path}: {variables[path]!r}')
         if sections is None or 'environment_variables' in sections:
             output['environment_variables'] = {}
             if self._returntype is ReturnType.TEXT:
@@ -1251,7 +1251,7 @@ def debug(self, sections=None):
                              'PATH'):
                 output['environment_variables'][variable] = os.getenv(variable)
                 if self._returntype is ReturnType.TEXT:
-                    print('%s: %r' % (variable, os.getenv(variable)))
+                    print(f'{variable}: {os.getenv(variable)!r}')
         return 0, output
 
     def log_bot_message(self, status, *args):

intelmq/bin/intelmqdump.py

@@ -92,24 +92,24 @@ def dump_info(fname, file_descriptor=None):
     else:
         try:
             if file_descriptor is None:
-                handle = open(fname, 'rt')
+                handle = open(fname)
                 fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
             else:
                 handle = file_descriptor
         except BlockingIOError:
             info = red('Dump file is locked.')
         except OSError as exc:
-            info = red('unable to open file: {!s}'.format(exc))
+            info = red(f'unable to open file: {exc!s}')
         else:
             try:
                 content = json.load(handle)
             except ValueError as exc:
-                info = red('unable to load JSON: {!s}'.format(exc))
+                info = red(f'unable to load JSON: {exc!s}')
             else:
                 try:
-                    info = "{!s} dumps".format(len(content.keys()))
+                    info = f"{len(content.keys())!s} dumps"
                 except AttributeError as exc:
-                    info = red("unable to count dumps: {!s}".format(exc))
+                    info = red(f"unable to count dumps: {exc!s}")
         finally:
             try:
                 if file_descriptor is None:
@@ -221,7 +221,7 @@ def main():
         filenames = [(fname, fname[len(DEFAULT_LOGGING_PATH):-5])
                      for fname in sorted(filenames)]
 
-        length = max([len(value[1]) for value in filenames])
+        length = max(len(value[1]) for value in filenames)
         print(bold("{c:>3}: {s:{length}} {i}".format(c='id', s='name (bot id)',
                                                      i='content',
                                                      length=length)))
@@ -249,7 +249,7 @@ def main():
         fname = os.path.join(DEFAULT_LOGGING_PATH, botid) + '.dump'
 
     if not os.path.isfile(fname):
-        print(bold('Given file does not exist: {}'.format(fname)))
+        print(bold(f'Given file does not exist: {fname}'))
         exit(1)
 
     answer = None
@@ -264,7 +264,7 @@ def main():
             info = dump_info(fname, file_descriptor=handle)
             handle.seek(0)
             available_answers = ACTIONS.keys()
-            print('Processing {}: {}'.format(bold(botid), info))
+            print(f'Processing {bold(botid)}: {info}')
 
             if info.startswith(str(red)):
                 available_opts = [item[0] for item in ACTIONS.values() if item[2]]
@@ -351,7 +351,7 @@ def main():
                                 print('Event converted to Report automatically.')
                                 msg = message.Report(message.MessageFactory.unserialize(msg)).serialize()
                         else:
-                            print(red("The given queue '{}' is not configured. Please retry with a valid queue.".format(queue_name)))
+                            print(red(f"The given queue '{queue_name}' is not configured. Please retry with a valid queue."))
                             break
                         try:
                             pipe.set_queues(queue_name, 'destination')
@@ -362,12 +362,12 @@ def main():
                                       ''.format(queue_name, traceback.format_exc())))
                         else:
                             del content[key]
-                            print(green('Recovered dump {}.'.format(i)))
+                            print(green(f'Recovered dump {i}.'))
                 finally:
                     save_file(handle, content)
                 if not content:
                     delete_file = True
-                    print('Deleting empty file {}'.format(fname))
+                    print(f'Deleting empty file {fname}')
                     break
             elif answer[0] == 'd':
                 # Delete entries or file
@@ -379,15 +379,15 @@ def main():
                 else:
                     # delete dumpfile
                     delete_file = True
-                    print('Deleting file {}'.format(fname))
+                    print(f'Deleting file {fname}')
                     break
             elif answer[0] == 's':
                 # Show entries by id
                 for count, (key, orig_value) in enumerate(content.items()):
                     value = copy.copy(orig_value)  # otherwise the raw field gets truncated
                     if count not in ids:
                         continue
-                    print('=' * 100, '\nShowing id {} {}\n'.format(count, key),
+                    print('=' * 100, f'\nShowing id {count} {key}\n',
                           '-' * 50)
                     if value.get('message_type') == 'base64':
                         if args.truncate and len(value['message']) > args.truncate:

intelmq/bin/intelmqsetup.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 """
 © 2019-2021 nic.at GmbH <intelmq-team@cert.at>
 

intelmq/bin/rewrite_config_files.py

@@ -43,6 +43,6 @@ def rewrite(fobj):
             with open(fn, 'r+') as f:
                 rewrite(f)
 
-    except IOError:
+    except OSError:
         traceback.print_exc()
         print('Could not open files. Wrong directory? Also see the --help.')

intelmq/bots/collectors/file/collector_file.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 # SPDX-FileCopyrightText: 2016 by Bundesamt für Sicherheit in der Informationstechnik
 #
 # SPDX-License-Identifier: AGPL-3.0-or-later

intelmq/bots/collectors/github_api/_collector_github_api.py

@@ -47,18 +47,18 @@ def process_request(self):
 
     def github_api(self, api_path: str, **kwargs) -> dict:
         try:
-            response = requests.get("{}".format(api_path), params=kwargs, headers=self.__user_headers)
+            response = requests.get(f"{api_path}", params=kwargs, headers=self.__user_headers)
             if response.status_code == 401:
                 # bad credentials
                 raise ValueError(response.json()['message'])
             else:
                 return response.json()
         except requests.RequestException:
-            raise ValueError("Unknown repository {!r}.".format(api_path))
+            raise ValueError(f"Unknown repository {api_path!r}.")
 
     @staticmethod
     def __produce_auth_header(username: str, password: str) -> dict:
-        encoded_auth_bytes = base64.b64encode(bytes('{}:{}'.format(username, password), encoding='utf-8'))
+        encoded_auth_bytes = base64.b64encode(bytes(f'{username}:{password}', encoding='utf-8'))
         return {
             'Authorization': 'Basic {}'.format(encoded_auth_bytes.decode('utf-8'))
         }

intelmq/bots/collectors/github_api/collector_github_contents_api.py

@@ -33,7 +33,7 @@ class GithubContentsAPICollectorBot(GithubAPICollectorBot):
     def init(self):
         super().init()
         if self.repository is not None:
-            self.__base_api_url = 'https://api.github.com/repos/{}/contents'.format(self.repository)
+            self.__base_api_url = f'https://api.github.com/repos/{self.repository}/contents'
         else:
             raise InvalidArgument('repository', expected='string')
 
@@ -82,7 +82,7 @@ def __recurse_repository_files(self, base_api_url: str, extracted_github_files:
                     if field_name in github_file:
                         extracted_github_file_data['extra'][field_name] = github_file[field_name]
                     else:
-                        self.logger.warning("Field '{}' does not exist in the Github file data.".format(field_name))
+                        self.logger.warning(f"Field '{field_name}' does not exist in the Github file data.")
                 extracted_github_files.append(extracted_github_file_data)
 
         return extracted_github_files

intelmq/bots/collectors/http/collector_http.py

@@ -40,7 +40,7 @@
     gnupg = None
 
 
-class Time(object):
+class Time:
     def __init__(self, delta=None):
         """ Delta is a datetime.timedelta JSON string, ex: '{"days"=-1}'. """
         self.time = datetime.now()
@@ -105,7 +105,7 @@ def process(self):
                 return
 
             if result.trust_level < 1:
-                self.logger.debug("Trust level not defined for key {}.".format(result.key_id))
+                self.logger.debug(f"Trust level not defined for key {result.key_id}.")
             elif result.trust_level < 3:
                 self.logger.debug("Low trust level for key {0.key_id}: {0.trust_level}.".format(result))
 
@@ -165,11 +165,11 @@ def verify_signature(self, data: bytes):
             http_url = self.signature_url
 
         # download signature file
-        self.logger.info("Downloading PGP signature from {}.".format(http_url))
+        self.logger.info(f"Downloading PGP signature from {http_url}.")
 
         resp = self.http_get(http_url)
         if resp.status_code // 100 != 2:
-            raise ValueError("Could not download PGP signature for report: {}.".format(resp.status_code))
+            raise ValueError(f"Could not download PGP signature for report: {resp.status_code}.")
 
         self.logger.info("PGP signature downloaded.")