Upgrade code with pyupgrade --py39-plus

Author: sebix

intelmq/bots/collectors/mail/collector_mail_body.py

@@ -6,7 +6,8 @@
 """
 Uses the common mail iteration method from the lib file.
 """
-from typing import Union, Iterable
+from typing import Union
+from collections.abc import Iterable
 
 from ._lib import MailCollectorBot
 

intelmq/bots/collectors/shodan/collector_stream.py

@@ -30,7 +30,7 @@
 class ShodanStreamCollectorBot(CollectorBot):
     "Collect the Shodan stream from the Shodan API"
     api_key: str = "<INSERT your API key>"
-    countries: List[str] = []
+    countries: list[str] = []
     alert: Optional[str] = None
 
     def init(self):

intelmq/bots/experts/gethostbyname/expert.py

@@ -33,7 +33,7 @@
 class GethostbynameExpertBot(ExpertBot):
     """Resolve the IP address for the FQDN"""
     fallback_to_url: bool = True
-    gaierrors_to_ignore: Tuple[int] = ()
+    gaierrors_to_ignore: tuple[int] = ()
     overwrite: bool = False
 
     def init(self):

intelmq/bots/experts/http/expert_status.py

@@ -24,7 +24,7 @@ class HttpStatusExpertBot(ExpertBot):
             Specifies if an existing 'status' value should be overwritten.
     """
     field: str = "source.url"  # The field containing the URL
-    success_status_codes: List[int] = []  # A list of status codes for success
+    success_status_codes: list[int] = []  # A list of status codes for success
     overwrite: bool = True
 
     def process(self):

intelmq/bots/experts/jinja/expert.py

@@ -27,8 +27,8 @@ class JinjaExpertBot(ExpertBot):
           extra.somejinjaoutput: file:///etc/intelmq/somejinjatemplate.j2
     """
 
-    fields: Dict[str, str] = {}
-    _templates: Dict[str, Union[str, Template]] = {}
+    fields: dict[str, str] = {}
+    _templates: dict[str, Union[str, Template]] = {}
     overwrite: bool = False
 
     def init(self):

intelmq/bots/experts/sieve/expert.py

@@ -173,7 +173,7 @@ def process(self) -> None:
 
     _date_op_map = {":before": operator.lt, ":after": operator.gt}
 
-    _cond_map: Dict[
+    _cond_map: dict[
         str,
         Callable[
             [

intelmq/bots/experts/threshold/expert.py

@@ -44,7 +44,8 @@
               messages seen (which will be the threshold value).
 
 """
-from typing import Iterable, Optional
+from typing import Optional
+from collections.abc import Iterable
 
 from intelmq.lib.bot import ExpertBot
 from intelmq.lib.exceptions import ConfigurationError

intelmq/bots/experts/url/expert.py

@@ -66,7 +66,7 @@ class URLExpertBot(ExpertBot):
     """
 
     overwrite: bool = False
-    skip_fields: Optional[List[str]] = None
+    skip_fields: Optional[list[str]] = None
 
     def init(self):
         if self.skip_fields is None:

intelmq/bots/experts/url2fqdn/expert.py

@@ -21,7 +21,7 @@ def init(self):
         warnings.warn(DEPRECATION_WARNING, DeprecationWarning)
 
     @staticmethod
-    def check(parameters: dict) -> Optional[List[List[str]]]:
+    def check(parameters: dict) -> Optional[list[list[str]]]:
         return [["warning", DEPRECATION_WARNING]]
 
     def process(self):

intelmq/bots/outputs/cif3/output.py

@@ -83,7 +83,7 @@ class CIF3OutputBot(OutputBot):
     add_feed_provider_as_tag: bool = False
     cif3_feed_confidence: float = 5
     cif3_static_confidence: bool = False
-    cif3_additional_tags: List[str] = []
+    cif3_additional_tags: list[str] = []
     cif3_token: Optional[str] = None
     cif3_url: Optional[str] = None
     fireball: int = 500

intelmq/bots/outputs/restapi/output.py

@@ -3,7 +3,7 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
 # -*- coding: utf-8 -*-
-from typing import Iterable
+from collections.abc import Iterable
 
 try:
     import requests

intelmq/bots/outputs/templated_smtp/output.py

@@ -112,7 +112,7 @@ class TemplatedSMTPOutputBot(OutputBot):
     username: Optional[str] = None
     password: Optional[str] = None
     verify_cert: bool = True
-    attachments: List[str] = []
+    attachments: list[str] = []
     mail_from: Optional[str] = None
     mail_to: Optional[str] = None
     subject: str = "IntelMQ event"

intelmq/bots/parsers/generic/parser_csv.py

@@ -19,7 +19,8 @@
 import csv
 import json
 import re
-from typing import Optional, Union, Iterable
+from typing import Optional, Union
+from collections.abc import Iterable
 
 from intelmq.lib import utils
 from intelmq.lib.bot import ParserBot

intelmq/bots/parsers/ioc_extractor/parser.py

@@ -23,7 +23,8 @@
 import re
 import pkg_resources
 
-from typing import Optional, Iterable
+from typing import Optional
+from collections.abc import Iterable
 
 from intelmq.lib.bot import ParserBot, utils
 from intelmq.lib.exceptions import InvalidArgument

intelmq/bots/parsers/shadowserver/_config.py

@@ -128,11 +128,11 @@ def enable_auto_update(enable):
     __config.auto_update = enable
 
 
-def get_feed_by_feedname(given_feedname: str) -> Optional[Tuple[str, Dict[str, Any]]]:
+def get_feed_by_feedname(given_feedname: str) -> Optional[tuple[str, dict[str, Any]]]:
     return __config.feedname_mapping.get(given_feedname, None)
 
 
-def get_feed_by_filename(given_filename: str) -> Optional[Tuple[str, Dict[str, Any]]]:
+def get_feed_by_filename(given_filename: str) -> Optional[tuple[str, dict[str, Any]]]:
     return __config.filename_mapping.get(given_filename, None)
 
 
@@ -164,7 +164,7 @@ def convert_float(value: str) -> Optional[float]:
     return float(value) if value else None
 
 
-def convert_http_host_and_url(value: str, row: Dict[str, str]) -> str:
+def convert_http_host_and_url(value: str, row: dict[str, str]) -> str:
     """
     URLs are split into hostname and path. The column names differ in reports.
     Compromised-Website: http_host, url
@@ -282,7 +282,7 @@ def scan_exchange_identifier(field):
     return 'vulnerable-exchange-server'
 
 
-def category_or_detail(value: str, row: Dict[str, str]) -> str:
+def category_or_detail(value: str, row: dict[str, str]) -> str:
     """
     Returns the category or detail field from the row.
     """

intelmq/bots/parsers/shodan/parser.py

@@ -460,7 +460,7 @@ def __repr__(self) -> str:
 }
 
 
-def _keys_conversion(x: Dict[str, Any]) -> List[str]:
+def _keys_conversion(x: dict[str, Any]) -> list[str]:
     '''
     extracts object keys to a list, for cases where the values they map to are empty/irrelevant
     '''
@@ -471,15 +471,15 @@ def _keys_conversion(x: Dict[str, Any]) -> List[str]:
 
 
 # in case item can be either T or List[T]
-def _maybe_single_to_list(x: Any) -> List[Any]:
+def _maybe_single_to_list(x: Any) -> list[Any]:
     '''
     converts non-list objects to lists with a single item and leaves lists as-is,
     used to harmonize fields which avoid lists when a single value is given
     '''
     return x if isinstance(x, list) else [x]
 
 
-def _dict_dict_to_obj_list(x: Dict[str, Dict[str, Any]], identifier: str = 'identifier') -> List[Dict[str, Any]]:
+def _dict_dict_to_obj_list(x: dict[str, dict[str, Any]], identifier: str = 'identifier') -> list[dict[str, Any]]:
     '''
     convert e.g
     {'OuterKey1': {'InnerKey1': 'Value1'}, 'OuterKey2': {'InnerKey2': 'Value2'}}
@@ -497,7 +497,7 @@ def _dict_dict_to_obj_list(x: Dict[str, Dict[str, Any]], identifier: str = 'iden
     return out
 
 
-def _get_first(variable: List[Any]) -> Any:
+def _get_first(variable: list[Any]) -> Any:
     '''
     get first element from list, if the list has any; raise NoValueException otherwise
     '''
@@ -507,7 +507,7 @@ def _get_first(variable: List[Any]) -> Any:
         raise NoValueException(f'empty list passed to _get_first')
 
 
-def _get_first_fqdn(variable: List[str]) -> str:
+def _get_first_fqdn(variable: list[str]) -> str:
     '''
     get first valid FQDN from a list of strings
     '''
@@ -519,7 +519,7 @@ def _get_first_fqdn(variable: List[str]) -> str:
     return first
 
 
-CONVERSIONS: Dict[str, Callable[[Any], Any]] = {
+CONVERSIONS: dict[str, Callable[[Any], Any]] = {
     'ftp.features': _dict_dict_to_obj_list,
     'timestamp': lambda x: x + '+00',
     'hostnames': _get_first_fqdn,
@@ -541,7 +541,7 @@ class ShodanParserBot(ParserBot):
     ignore_errors = True
     minimal_mode = False
 
-    def apply_mapping(self, mapping: Dict[str, Any], data: Dict[str, Any], key_path: Tuple[str, ...] = ()) -> Dict[str, Any]:
+    def apply_mapping(self, mapping: dict[str, Any], data: dict[str, Any], key_path: tuple[str, ...] = ()) -> dict[str, Any]:
         self.logger.debug(f'Applying mapping {mapping!r} to data {data!r}.')
         event = {}
         for key in data.keys() & mapping.keys():

intelmq/bots/parsers/twitter/parser.py

@@ -20,7 +20,7 @@ def init(self):
         super().init()
 
     @staticmethod
-    def check(parameters: dict) -> Optional[List[List[str]]]:
+    def check(parameters: dict) -> Optional[list[list[str]]]:
         return [["warning", DEPRECATION_WARNING]]
 
 

intelmq/lib/bot.py

@@ -56,7 +56,7 @@ class Bot:
     __stats_cache: cache.Cache = None
     __source_pipeline = None
     __destination_pipeline = None
-    __log_buffer: List[tuple] = []
+    __log_buffer: list[tuple] = []
     # runtime_file
     __runtime_settings: Optional[dict] = None
     # settings provided via parameter
@@ -612,7 +612,7 @@ def __print_log_buffer(self):
                 print(level.upper(), '-', message)
         self.__log_buffer = []
 
-    def __check_bot_id(self, name: str) -> Tuple[str, str, str]:
+    def __check_bot_id(self, name: str) -> tuple[str, str, str]:
         res = re.fullmatch(r'([0-9a-zA-Z\-]+)(\.[0-9]+)?', name)
         if res:
             if not (res.group(2) and threading.current_thread() == threading.main_thread()):
@@ -975,7 +975,7 @@ def set_request_parameters(self):
         self.http_header['User-agent'] = self.http_user_agent
 
     @staticmethod
-    def check(parameters: dict) -> Optional[List[List[str]]]:
+    def check(parameters: dict) -> Optional[list[list[str]]]:
         """
         The bot's own check function can perform individual checks on it's
         parameters.

intelmq/lib/datatypes.py

@@ -113,7 +113,7 @@ def validate(value: str) -> [Callable, Optional[str]]:
         :return: correct time conversion function and the format string
         """
 
-        split_value: List[str] = value.split('|')
+        split_value: list[str] = value.split('|')
         conversion: Callable
         conversion_name: str = split_value[0]
         format_string: Optional[str] = split_value[1] if len(split_value) > 1 else None
@@ -139,19 +139,4 @@ def validate(value: str) -> [Callable, Optional[str]]:
         return conversion, format_string
 
 
-if version_info < (3, 9):
-    class Dict39(dict):
-        """
-        Python 3.9 introduced the handy | operator for dicts.
-        For backwards-compatibility, this is the backport
-        as IntelMQ supports Python >= 3.7
-        """
-        def __or__(self, other: dict) -> 'Dict39':
-            """
-            Create a new dictionary with the merged keys and values of d and other, which must both be dictionaries. The values of other take priority when d and other share keys.
-            """
-            ret = Dict39(self.copy())
-            ret.update(other)
-            return ret
-else:
-    Dict39 = dict
+Dict39 = dict

intelmq/lib/message.py

@@ -13,7 +13,8 @@
 import re
 import warnings
 from collections import defaultdict
-from typing import Any, Dict, Iterable, Optional, Sequence, Union, Tuple
+from typing import Any, Dict, Optional, Union, Tuple
+from collections.abc import Iterable, Sequence
 from pkg_resources import resource_filename
 
 import intelmq.lib.exceptions as exceptions
@@ -332,7 +333,7 @@ def unserialize(message_string: str):
         message = json.loads(message_string)
         return message
 
-    def __is_valid_key(self, key: str) -> Tuple[bool, str]:
+    def __is_valid_key(self, key: str) -> tuple[bool, str]:
         try:
             class_name, subitem = self.__get_type_config(key)
         except KeyError:

intelmq/lib/mixins/stomp.py

@@ -63,7 +63,7 @@ class StompMixin:
     # Helper methods intended to be used in subclasses
 
     @classmethod
-    def stomp_bot_parameters_check(cls, parameters: dict) -> List[List[str]]:
+    def stomp_bot_parameters_check(cls, parameters: dict) -> list[list[str]]:
         """Intended to be used in bots' `check()` static/class method."""
         logs = []
         cls.__verify_parameters(
@@ -80,7 +80,7 @@ def stomp_bot_runtime_initial_check(self) -> None:
             on_error=self.__raise_value_error,
         )
 
-    def prepare_stomp_connection(self) -> Tuple['stomp.Connection', dict]:
+    def prepare_stomp_connection(self) -> tuple['stomp.Connection', dict]:
         """
         Get a `(<STOMP connection>, <STOMP connect arguments>)` pair.
 
@@ -178,7 +178,7 @@ def __get_own_attribute(self, param_name: str) -> Any:
     def __raise_value_error(self, msg: str) -> NoReturn:
         raise ValueError(msg)
 
-    def __get_ssl_and_connect_kwargs(self) -> Tuple[dict, dict]:
+    def __get_ssl_and_connect_kwargs(self) -> tuple[dict, dict]:
         # Note: a *non-empty* and *non-None* `ca_certs` argument must
         # always be passed to `set_ssl()`; otherwise the `stomp.py`'s
         # machinery would *not* enable any certificate verification!
@@ -276,7 +276,7 @@ def patch_stomp_transport_ssl(cls) -> None:
     #
     # Proxying/substituting `ssl` tools for `stomp.transport` module
 
-    def __dir__(self) -> List[str]:
+    def __dir__(self) -> list[str]:
         return dir(ssl)
 
     def __getattribute__(self, name: str) -> Any:

intelmq/lib/processmanager.py

@@ -15,7 +15,8 @@
 import sys
 import time
 import xmlrpc.client
-from typing import Union, Iterable
+from typing import Union
+from collections.abc import Iterable
 
 
 from intelmq import (DEFAULT_LOGGING_LEVEL,  # noqa: F401

intelmq/lib/splitreports.py

@@ -39,12 +39,13 @@
    chunk size must take this into account, but multiplying the actual
    limit by 3/4 and subtracting a generous amount for the meta data.
 """
-from typing import BinaryIO, Generator, List, Optional
+from typing import BinaryIO, List, Optional
+from collections.abc import Generator
 
 from intelmq.lib.message import Report
 
 
-def split_chunks(chunk: bytes, chunk_size: int) -> List[bytes]:
+def split_chunks(chunk: bytes, chunk_size: int) -> list[bytes]:
     """Split a bytestring into chunk_size pieces at ASCII newlines characters.
 
     The return value is a list of bytestring objects. Appending all of