Updated linter configs

boefjes/boefjes/katalogus/dependencies/plugins.py

@@ -1,3 +1,4 @@
+import contextlib
 import logging
 from pathlib import Path
 from typing import Dict, Iterable, Iterator, List, Optional
@@ -201,10 +202,8 @@ def _assert_settings_match_schema(self, organisation_id: str, plugin_id: str):
                 raise SettingsNotConformingToSchema(organisation_id, plugin_id, e.message) from e
 
     def _set_plugin_enabled(self, plugin: PluginType, organisation_id: str) -> PluginType:
-        try:
+        with contextlib.suppress(KeyError, NotFound):
             plugin.enabled = self.plugin_enabled_store.get_by_id(plugin.id, plugin.repository_id, organisation_id)
-        except (KeyError, NotFound):
-            pass
 
         return plugin
 

boefjes/boefjes/katalogus/tests/integration/test_api.py

@@ -57,7 +57,7 @@ def setUp(self) -> None:
     def tearDown(self) -> None:
         session = sessionmaker(bind=get_engine())()
 
-        for table in SQL_BASE.metadata.tables.keys():
+        for table in SQL_BASE.metadata.tables:
             session.execute(f"DELETE FROM {table} CASCADE")
 
         session.commit()

boefjes/boefjes/katalogus/tests/integration/test_sql_repositories.py

@@ -46,7 +46,7 @@ def setUp(self) -> None:
     def tearDown(self) -> None:
         session = sessionmaker(bind=get_engine())()
 
-        for table in SQL_BASE.metadata.tables.keys():
+        for table in SQL_BASE.metadata.tables:
             session.execute(f"DELETE FROM {table} CASCADE")
 
         session.commit()
@@ -166,9 +166,8 @@ def test_settings_storage(self):
 
         self.assertEqual(dict(), settings_storage.get_all(org.id, plugin_id))
 
-        with self.assertRaises(StorageError):
-            with self.settings_storage as settings_storage:
-                settings_storage.create("TEST_SETTING", "123.9", organisation_id, 65 * "a")
+        with self.assertRaises(StorageError), self.settings_storage as settings_storage:
+            settings_storage.create("TEST_SETTING", "123.9", organisation_id, 65 * "a")
 
     def test_settings_storage_values_field_limits(self):
         organisation_id = "test"

boefjes/boefjes/plugin_repository/api/routers/simplestreams.py

@@ -23,7 +23,7 @@ def index_file():
                 "datatype": "image-downloads",
                 "path": "streams/v1/images.json",
                 "format": "products:1.0",
-                "products": [name for name in index.images.keys()],
+                "products": [name for name in index.images],
             }
         }
     }
@@ -63,9 +63,8 @@ def images_file():
                 "sha256": file.hash,
                 "path": f"{BASE_URL}/images/{file.location.relative_to(PLUGINS_DIR)}",
             }
-            if isinstance(file, CombinedFile):
-                if file.combined_squashfs_sha256 is not None:
-                    version["combined_squashfs_sha256"] = file.combined_squashfs_sha256
+            if isinstance(file, CombinedFile) and file.combined_squashfs_sha256 is not None:
+                version["combined_squashfs_sha256"] = file.combined_squashfs_sha256
             versions[created]["items"][file.location.name] = version
 
         content = {

boefjes/boefjes/plugins/kat_binaryedge/databases/normalize.py

@@ -66,28 +66,24 @@ def run(normalizer_meta: NormalizerMeta, raw: Union[bytes, str]) -> Iterable[OOI
         yield ip_port_ooi
 
         software_version = None
+        data = scan.get("result", {}).get("data", {})
         if module == "cassandra":
-            for cluster in scan.get("result", {}).get("data", {}).get("cluster", []):
+            for cluster in data.get("cluster", []):
                 if "cassandraVersion" in cluster:
                     software_version = cluster["cassandraVersion"]
-        elif module == "elasticsearch" or module == "memcached":
-            if "version" in scan.get("result", {}).get("data", {}):
-                software_version = scan["result"]["data"]["version"]
+                    break
+        elif module == "elasticsearch" or module == "memcached" and "version" in data:
+            software_version = data["version"]
             # TODO: jvm.version, jvm.vm_version, jvm.vm_vendor
-        elif module == "mongodb":
-            if "version" in scan.get("result", {}).get("data", {}).get("serverInfo"):
-                software_version = scan["result"]["data"]["serverInfo"]["version"]
+        elif module == "mongodb" and "version" in data.get("serverInfo", {}):
+            software_version = data["serverInfo"]["version"]
             # TODO: 'serverInfo.OpenSSLVersion, scan['result']['data']['serverInfo']['openssl']{running,compiled}
             # TODO: buildEnvironment.cc
-        elif module == "redis":
-            if "redis_version" in scan.get("result", {}).get("data", {}):
-                software_version = scan["result"]["data"]["redis_version"]
+        elif module == "redis" and "redis_version" in data:
+            software_version = data["redis_version"]
             # TODO: data.gccversion
 
-        if software_version:
-            software_ooi = Software(name=module, version=software_version)
-        else:
-            software_ooi = Software(name=module)
+        software_ooi = Software(name=module, version=software_version) if software_version else Software(name=module)
         yield software_ooi
         software_instance_ooi = SoftwareInstance(ooi=ip_port_ooi.reference, software=software_ooi.reference)
         yield software_instance_ooi

boefjes/boefjes/plugins/kat_binaryedge/protocols/normalize.py

@@ -116,16 +116,11 @@ def run(normalizer_meta: NormalizerMeta, raw: Union[bytes, str]) -> Iterable[OOI
             )
         if "robot_result_enum" in vulns.get("robot", {}):
             robot = vulns["robot"]["robot_result_enum"]
-            if robot == "VULNERABLE_WEAK_ORACLE":
-                # FIXME: new KAT-Finding (low)?
-                pass  # The server is vulnerable but the attack would take too long
-            elif robot == "VULNERABLE_STRONG_ORACLE":
-                # FIXME: new KAT-Finding (high)?
-                pass  # The server is vulnerable and real attacks are feasible
-            elif robot == "NOT_VULNERABLE_NO_ORACLE":
-                pass  # The server supports RSA cipher suites but does not act as an oracle
-            elif robot == "NOT_VULNERABLE_RSA_NOT_SUPPORTED":
-                pass  # The server does not supports RSA cipher suites
-            elif robot == "UNKNOWN_INCONSISTENT_RESULTS":
-                # FIXME: KATFinding (low)?
-                pass  # Could not determine whether the server is vulnerable or not
+            if robot in (
+                "VULNERABLE_WEAK_ORACLE",  # the server is vulnerable but the attack would take too long
+                "VULNERABLE_STRONG_ORACLE",  # the server is vulnerable and real attacks are feasible
+                "NOT_VULNERABLE_NO_ORACLE",  # the server supports RSA cipher suites but does not act as an oracle
+                "NOT_VULNERABLE_RSA_NOT_SUPPORTED",  # the server does not supports RSA cipher suites
+                "UNKNOWN_INCONSISTENT_RESULTS",  # could not determine whether the server is vulnerable or not
+            ):
+                pass  # todo

boefjes/boefjes/plugins/kat_dns/normalize.py

@@ -163,11 +163,10 @@ def register_record(record: DNSRecord) -> DNSRecord:
 
     # DKIM
     dkim_results = results["dkim_response"]
-    if dkim_results not in ["NXDOMAIN", "Timeout"]:
-        if "rcode NOERROR" == dkim_results.split("\n")[2]:
-            yield DKIMExists(
-                hostname=input_hostname.reference,
-            )
+    if dkim_results not in ["NXDOMAIN", "Timeout"] and dkim_results.split("\n")[2] == "rcode NOERROR":
+        yield DKIMExists(
+            hostname=input_hostname.reference,
+        )
 
     # DMARC
     dmarc_results = results["dmarc_response"]

boefjes/boefjes/plugins/kat_fierce/fierce.py

@@ -7,6 +7,7 @@
 
 import argparse
 import concurrent.futures
+import contextlib
 import functools
 import http.client
 import ipaddress
@@ -226,7 +227,8 @@ def get_stripped_file_lines(filename):
     Return lines of a file with whitespace removed
     """
     try:
-        lines = open(filename).readlines()
+        with open(filename) as f:
+            lines = f.readlines()
     except FileNotFoundError:
         fatal(f"Could not open file: {filename!r}")
 
@@ -290,10 +292,7 @@ def fierce(**kwargs):
 
     ns = recursive_query(resolver, domain, "NS", tcp=kwargs["tcp"])
 
-    if ns:
-        domain_name_servers = [n.to_text() for n in ns]
-    else:
-        domain_name_servers = []
+    domain_name_servers = [n.to_text() for n in ns] if ns else []
 
     output["NS"] = domain_name_servers if ns else "failure"
 
@@ -432,10 +431,8 @@ def parse_args(args):
 def main():
     args = parse_args(sys.argv[1:])
 
-    try:
+    with contextlib.suppress(KeyboardInterrupt):
         fierce(**vars(args))
-    except KeyboardInterrupt:
-        pass
 
 
 if __name__ == "__main__":

boefjes/boefjes/plugins/kat_leakix/normalize.py

@@ -48,10 +48,7 @@ def run(normalizer_meta: NormalizerMeta, raw: Union[bytes, str]) -> Iterable_[OO
         as_number = event["network"]["asn"]
         as_name = event["network"]["organization_name"]
         if as_number:
-            if as_name:
-                as_ooi = AutonomousSystem(number=as_number, name=as_name)
-            else:
-                as_ooi = AutonomousSystem(number=as_number)
+            as_ooi = AutonomousSystem(number=as_number, name=as_name) if as_name else AutonomousSystem(number=as_number)
             yield as_ooi
 
         if ip:
@@ -183,10 +180,7 @@ def run(normalizer_meta: NormalizerMeta, raw: Union[bytes, str]) -> Iterable_[OO
             for tag in event.get("tags", {}):
                 if re.match("cve-[0-9]{4}-[0-9]{4,6}", tag):
                     ft = CVEFindingType(id=tag)
-                    if software_ooi:
-                        cve_ooi = software_ooi
-                    else:
-                        cve_ooi = ip_port_ooi
+                    cve_ooi = software_ooi if software_ooi else ip_port_ooi
                     f = Finding(finding_type=ft.reference, ooi=cve_ooi.reference)
                     yield ft
                     yield f

boefjes/boefjes/plugins/kat_snyk/check_version.py

@@ -41,7 +41,7 @@ def check_version(version1: str, version2: str) -> VersionCheck:
         if pack1[0].isnumeric() and pack2[0].isnumeric():
             # Has a package-name, but it starts with version-numbers  # https://snyk.io/vuln/debian%3A12%3Awordpress
             first_part_check = check_version(pack1[0], pack2[0])
-            if not first_part_check == VersionCheck.EQUAL:
+            if first_part_check != VersionCheck.EQUAL:
                 return first_part_check
             else:
                 # Version is the same, but package different.. impossible to compare
@@ -96,10 +96,7 @@ def check_version_agains_versionlist(my_version: str, all_versions: List[str]):
                 "Unexpected input, missing closing bracket for %s,%s. Ignoring input.", lowerbound, upperbound
             )
             return False, None
-        if lowerbound[0] == "(":
-            lowerbound_versioncheck = VersionCheck.GREATER
-        else:
-            lowerbound_versioncheck = VersionCheck.GREATER_EQUAL
+        lowerbound_versioncheck = VersionCheck.GREATER if lowerbound[0] == "(" else VersionCheck.GREATER_EQUAL
         lowerbound = lowerbound[1:].strip()
         if len(lowerbound) == 0:
             # Example: "(,1.2)"  # https://snyk.io/vuln/maven%3Aorg.apache.nifi%3Anifi-security-utils
@@ -144,10 +141,7 @@ def check_version_agains_versionlist(my_version: str, all_versions: List[str]):
     upperbound_versioncheck = None
     if end_bracket:
         # Example: "(1.2,1.4]"
-        if upperbound[-1] == ")":
-            upperbound_versioncheck = VersionCheck.SMALLER
-        else:
-            upperbound_versioncheck = VersionCheck.SMALLER_EQUAL
+        upperbound_versioncheck = VersionCheck.SMALLER if upperbound[-1] == ")" else VersionCheck.SMALLER_EQUAL
         upperbound = upperbound[:-1].strip()
     elif start_inequality:
         # Example: "<=1.4"

boefjes/boefjes/plugins/kat_webpage_analysis/boefje.json

@@ -8,6 +8,6 @@
         "produces": [
             "HTTPHeader"
         ],
-        "environment_keys": [],
+        "environment_keys": ["USERAGENT"],
         "scan_level": 2
     }

boefjes/boefjes/plugins/kat_webpage_analysis/main.py

@@ -16,7 +16,7 @@
 
 def run(boefje_meta: BoefjeMeta) -> List[Tuple[set, Union[bytes, str]]]:
     input_ = boefje_meta.arguments["input"]
-    useragent = getenv("useragent", default="OpenKAT")
+    useragent = getenv("USERAGENT", default="OpenKAT")
 
     uri = get_uri(input_)
     ip = input_["website"]["ip_service"]["ip_port"]["address"]["address"]

boefjes/boefjes/plugins/kat_wpscan/main.py

@@ -13,7 +13,7 @@ def run(boefje_meta: BoefjeMeta) -> List[Tuple[set, Union[bytes, str]]]:
     client = docker.from_env()
     input_ = boefje_meta.arguments["input"]
 
-    if not input_["software"]["name"] == "WordPress" or (
+    if input_["software"]["name"] != "WordPress" or (
         "netloc" not in input_["ooi"] or "name" not in input_["ooi"]["netloc"]
     ):
         return [(set(), "")]
@@ -22,7 +22,7 @@ def run(boefje_meta: BoefjeMeta) -> List[Tuple[set, Union[bytes, str]]]:
     path = input_["ooi"]["path"]
     scheme = input_["ooi"]["scheme"]
 
-    if not scheme == "https":
+    if scheme != "https":
         return [(set(), "")]
 
     url = f"{scheme}://{hostname}{path}"

bytes/bytes/api/router.py

@@ -149,10 +149,7 @@ async def create_raw(
     meta_repository: MetaDataRepository = Depends(create_meta_data_repository),
     event_manager: EventManager = Depends(create_event_manager),
 ) -> RawResponse:
-    if mime_types is None:
-        parsed_mime_types = []
-    else:
-        parsed_mime_types = [MimeType(value=mime_type) for mime_type in mime_types]
+    parsed_mime_types = [] if mime_types is None else [MimeType(value=mime_type) for mime_type in mime_types]
 
     try:
         meta = meta_repository.get_boefje_meta_by_id(boefje_meta_id)
@@ -206,10 +203,7 @@ def get_raws(
 ) -> List[RawDataMeta]:
     """Get a filtered list of RawDataMeta objects, which contains metadata of a RawData object without the contents"""
 
-    if mime_types is None:
-        parsed_mime_types = []
-    else:
-        parsed_mime_types = [MimeType(value=mime_type) for mime_type in mime_types]
+    parsed_mime_types = [] if mime_types is None else [MimeType(value=mime_type) for mime_type in mime_types]
 
     query_filter = RawDataFilter(
         organization=organization,

bytes/tests/conftest.py

@@ -75,7 +75,7 @@ def meta_repository(
     session.commit()
 
     sessionmaker(bind=engine, autocommit=True)().execute(
-        ";".join([f"TRUNCATE TABLE {t} CASCADE" for t in SQL_BASE.metadata.tables.keys()])
+        ";".join([f"TRUNCATE TABLE {t} CASCADE" for t in SQL_BASE.metadata.tables])
     )
 
 
@@ -91,7 +91,7 @@ def bytes_api_client(settings) -> Iterator[BytesAPIClient]:
     )
 
     sessionmaker(bind=get_engine(settings.bytes_db_uri), autocommit=True)().execute(
-        ";".join([f"TRUNCATE TABLE {t} CASCADE" for t in SQL_BASE.metadata.tables.keys()])
+        ";".join([f"TRUNCATE TABLE {t} CASCADE" for t in SQL_BASE.metadata.tables])
     )
 
 

bytes/tests/integration/test_meta_repository.py

@@ -70,11 +70,10 @@ def test_boefje_id_length(meta_repository: SQLMetaDataRepository) -> None:
         boefje_meta.boefje.id = 64 * "a"
         meta_repository.save_boefje_meta(boefje_meta)
 
-    with pytest.raises(DataError):
-        with meta_repository:
-            boefje_meta.id = str(uuid.uuid4())
-            boefje_meta.boefje.id = 65 * "a"
-            meta_repository.save_boefje_meta(boefje_meta)
+    with pytest.raises(DataError), meta_repository:
+        boefje_meta.id = str(uuid.uuid4())
+        boefje_meta.boefje.id = 65 * "a"
+        meta_repository.save_boefje_meta(boefje_meta)
 
     meta_repository.session.rollback()  # make sure to roll back the session, so we can clean up the db
 
@@ -86,11 +85,10 @@ def test_boefje_organization_id_length(meta_repository: SQLMetaDataRepository) -
         boefje_meta.organization = 32 * "t"
         meta_repository.save_boefje_meta(boefje_meta)
 
-    with pytest.raises(DataError):
-        with meta_repository:
-            boefje_meta.id = str(uuid.uuid4())
-            boefje_meta.organization = 33 * "t"
-            meta_repository.save_boefje_meta(boefje_meta)
+    with pytest.raises(DataError), meta_repository:
+        boefje_meta.id = str(uuid.uuid4())
+        boefje_meta.organization = 33 * "t"
+        meta_repository.save_boefje_meta(boefje_meta)
 
     meta_repository.session.rollback()  # make sure to roll back the session, so we can clean up the db
 
@@ -262,11 +260,10 @@ def test_normalizer_id_length(meta_repository: SQLMetaDataRepository) -> None:
         normalizer_meta.normalizer.id = 64 * "a"
         meta_repository.save_normalizer_meta(normalizer_meta)
 
-    with pytest.raises(DataError):
-        with meta_repository:
-            normalizer_meta.id = str(uuid.uuid4())
-            normalizer_meta.normalizer.id = 65 * "a"
-            meta_repository.save_normalizer_meta(normalizer_meta)
+    with pytest.raises(DataError), meta_repository:
+        normalizer_meta.id = str(uuid.uuid4())
+        normalizer_meta.normalizer.id = 65 * "a"
+        meta_repository.save_normalizer_meta(normalizer_meta)
 
     meta_repository.session.rollback()  # make sure to roll back the session, so we can clean up the db