Add typehints and make mypy happy

.pre-commit-config.yaml

@@ -48,8 +48,13 @@ repos:
     hooks:
     -   id: black
 
-- repo: https://github.com/charliermarsh/ruff-pre-commit
-  rev: 'v0.0.275'
-  hooks:
-    - id: ruff
-      args: [ --fix, --exit-non-zero-on-fix ]
+-   repo: https://github.com/charliermarsh/ruff-pre-commit
+    rev: 'v0.0.275'
+    hooks:
+    -   id: ruff
+        args: [ --fix, --exit-non-zero-on-fix ]
+
+-   repo: https://github.com/pre-commit/mirrors-mypy
+    rev: "v1.5.1"
+    hooks:
+      - id: mypy

docsrc/conf.py

@@ -8,6 +8,7 @@
 
 import sys
 from pathlib import Path
+from unittest import mock
 
 FACT_SRC = Path(__file__).parent.parent / 'src'
 sys.path.insert(0, str(FACT_SRC))
@@ -119,5 +120,15 @@
 autodoc_typehints = 'description'
 
 
+class BaseMock:
+    metadata = None
+
+
+orm_mock = mock.Mock()
+orm_mock.DeclarativeBase = BaseMock
+sys.modules['sqlalchemy.orm'] = orm_mock
+sys.modules['sqlalchemy.orm.exc'] = mock.Mock()
+
+
 def setup(app):
     app.add_css_file('css/custom.css')

pyproject.toml

@@ -94,3 +94,24 @@ fixture-parentheses = false
 inline-quotes = "single"
 multiline-quotes = "single"
 docstring-quotes = "double"
+
+[tool.mypy]
+check_untyped_defs = true
+enable_error_code = "ignore-without-code"
+exclude = [
+    "bin/",
+    "docker/",
+    "install/venv",
+]
+follow_imports = "silent"
+ignore_missing_imports = true
+install_types = true
+no_error_summary = true
+non_interactive = true
+pretty = true
+show_error_codes = true
+show_error_context = true
+strict_equality = true
+warn_redundant_casts = true
+warn_unreachable = true
+warn_unused_ignores = true

src/analysis/PluginBase.py

@@ -7,8 +7,7 @@
 from queue import Empty
 from time import time
 
-from packaging.version import InvalidVersion
-from packaging.version import parse as parse_version
+from packaging.version import InvalidVersion, parse as parse_version
 
 import config
 from helperFunctions.process import (
@@ -20,10 +19,11 @@
 )
 from helperFunctions.tag import TagColor
 from plugins.base import BasePlugin
-from typing import TYPE_CHECKING
+from typing import Iterable, TYPE_CHECKING
 
 if TYPE_CHECKING:
     from objects.file import FileObject
+    from helperFunctions.types import MpValue, MpArray
 
 META_KEYS = {
     'tags',
@@ -62,35 +62,33 @@ class AnalysisBasePlugin(BasePlugin):
     This is the base plugin. All analysis plugins should be a subclass of this class.
     """
 
-    # must be set by the plugin:
-    FILE = None
-    NAME = None
-    DESCRIPTION = None
-    VERSION = None
+    # must be set by the plugin: NAME, FILE from BasePlugin and:
+    DESCRIPTION: str = ''
+    VERSION: str = ''
 
-    # can be set by the plugin:
-    RECURSIVE = True  # If `True` (default) recursively analyze included files
-    TIMEOUT = 300
-    SYSTEM_VERSION = None
-    MIME_BLACKLIST = []  # noqa: RUF012
-    MIME_WHITELIST = []  # noqa: RUF012
+    # can be set by the plugin: DEPENDENCIES from BasePlugin and:
+    RECURSIVE: bool = True  # If `True` (default) recursively analyze included files
+    TIMEOUT: int = 300
+    SYSTEM_VERSION: str | None = None
+    MIME_BLACKLIST: Iterable[str] = ()
+    MIME_WHITELIST: Iterable[str] = ()
 
     ANALYSIS_STATS_LIMIT = 1000
 
     def __init__(self, no_multithread=False, view_updater=None):
         super().__init__(plugin_path=self.FILE, view_updater=view_updater)
         self._check_plugin_attributes()
         self.additional_setup()
-        self.in_queue = Queue()
-        self.out_queue = Queue()
-        self.stop_condition = Value('i', 0)
+        self.in_queue: Queue[FileObject] = Queue()
+        self.out_queue: Queue[FileObject] = Queue()
+        self.stop_condition: MpValue[int] = Value('i', 0)  # type: ignore[assignment]
         self.workers = []
         self.thread_count = 1 if no_multithread else self._get_thread_count()
-        self.active = [Value('i', 0) for _ in range(self.thread_count)]
+        self.active: list[MpValue[int]] = [Value('i', 0) for _ in range(self.thread_count)]  # type: ignore[misc]
         self.manager = Manager()
-        self.analysis_stats = Array(ctypes.c_float, self.ANALYSIS_STATS_LIMIT)
-        self.analysis_stats_count = Value('i', 0)
-        self.analysis_stats_index = Value('i', 0)
+        self.analysis_stats: MpArray[ctypes.c_float] = Array(ctypes.c_float, self.ANALYSIS_STATS_LIMIT)
+        self.analysis_stats_count: MpValue[int] = Value('i', 0)  # type: ignore[assignment]
+        self.analysis_stats_index: MpValue[int] = Value('i', 0)  # type: ignore[assignment]
 
     def _get_thread_count(self):
         """
@@ -124,7 +122,7 @@ def shutdown(self):
 
     def _check_plugin_attributes(self):
         for attribute in ['FILE', 'NAME', 'VERSION']:
-            if getattr(self, attribute, None) is None:
+            if not bool(getattr(self, attribute, None)):
                 raise PluginInitException(f'Plugin {self.NAME} is missing {attribute} in configuration', plugin=self)
         self._check_version(self.VERSION)
         if self.SYSTEM_VERSION:
@@ -229,9 +227,9 @@ def worker_processing_with_timeout(self, worker_id, next_task: FileObject):
         result_fo.processed_analysis[self.NAME] = sanitize_processed_analysis(processed_analysis_entry)
         self.out_queue.put(result_fo)
 
-    def _update_duration_stats(self, duration):
+    def _update_duration_stats(self, duration: float):
         with self.analysis_stats.get_lock():
-            self.analysis_stats[self.analysis_stats_index.value] = duration
+            self.analysis_stats[self.analysis_stats_index.value] = ctypes.c_float(duration)
         self.analysis_stats_index.value += 1
         if self.analysis_stats_index.value >= self.ANALYSIS_STATS_LIMIT:
             # if the stats array is full, overwrite the oldest result

src/analysis/YaraPluginBase.py

@@ -4,13 +4,18 @@
 import re
 import subprocess
 from pathlib import Path
+from shlex import split
 
 import yaml
 from yaml.parser import ParserError
 
 from analysis.PluginBase import AnalysisBasePlugin, PluginInitException
 from helperFunctions.fileSystem import get_src_dir
 
+MATCH_REGEX = re.compile(r'((0x[a-f0-9]*):(\$[a-zA-Z0-9_]+):\s(.+))+')
+SPLIT_REGEX = re.compile(r'\n*.*\[.*]\s/.+\n*')
+RULE_REGEX = re.compile(r'(\w*)\s\[(.*)]\s([.]{0,2}/)(.+)')
+
 
 class YaraBasePlugin(AnalysisBasePlugin):
     """
@@ -20,7 +25,6 @@ class YaraBasePlugin(AnalysisBasePlugin):
     NAME = 'Yara_Base_Plugin'
     DESCRIPTION = 'this is a Yara plugin'
     VERSION = '0.0'
-    FILE = None
 
     def __init__(self, view_updater=None):
         """
@@ -37,8 +41,10 @@ def __init__(self, view_updater=None):
         super().__init__(view_updater=view_updater)
 
     def get_yara_system_version(self):
-        with subprocess.Popen(['yara', '--version'], stdout=subprocess.PIPE) as process:
-            yara_version = process.stdout.readline().decode().strip()
+        process = subprocess.run(split('yara --version'), capture_output=True, text=True)
+        if process.returncode != 0:
+            raise RuntimeError('Could not determine YARA version. Is YARA installed correctly?')
+        yara_version = process.stdout.strip()
 
         access_time = int(Path(self.signature_path).stat().st_mtime)
         return f'{yara_version}-{access_time}'
@@ -47,10 +53,9 @@ def process_object(self, file_object):
         if self.signature_path is not None:
             compiled_flag = '-C' if Path(self.signature_path).read_bytes().startswith(b'YARA') else ''
             command = f'yara {compiled_flag} --print-meta --print-strings {self.signature_path} {file_object.file_path}'
-            with subprocess.Popen(command, shell=True, stdout=subprocess.PIPE) as process:
-                output = process.stdout.read().decode()
+            process = subprocess.run(split(command), capture_output=True, text=True)
             try:
-                result = self._parse_yara_output(output)
+                result = self._parse_yara_output(process.stdout)
                 file_object.processed_analysis[self.NAME] = result
                 file_object.processed_analysis[self.NAME]['summary'] = list(result.keys())
             except (ValueError, TypeError):
@@ -68,42 +73,38 @@ def _get_signature_file(self, plugin_path):
         return str(Path(get_src_dir()) / 'analysis/signatures' / sig_file_name)
 
     @staticmethod
-    def _parse_yara_output(output):
-        resulting_matches = {}
-
+    def _parse_yara_output(output) -> dict[str, dict]:
         match_blocks, rules = _split_output_in_rules_and_matches(output)
 
-        matches_regex = re.compile(r'((0x[a-f0-9]*):(\$[a-zA-Z0-9_]+):\s(.+))+')
+        resulting_matches: dict[str, dict] = {}
         for index, rule in enumerate(rules):
-            for match in matches_regex.findall(match_blocks[index]):
-                _append_match_to_result(match, resulting_matches, rule)
+            rule_name, meta_string, _, _ = rule
+            for _, offset, matched_tag, matched_string in MATCH_REGEX.findall(match_blocks[index]):
+                resulting_matches.setdefault(
+                    rule_name,
+                    {
+                        'rule': rule_name,
+                        'matches': True,
+                        'strings': [],
+                        'meta': _parse_meta_data(meta_string),
+                    },
+                )['strings'].append((int(offset, 16), matched_tag, matched_string))
 
         return resulting_matches
 
 
 def _split_output_in_rules_and_matches(output):
-    split_regex = re.compile(r'\n*.*\[.*\]\s/.+\n*')
-    match_blocks = split_regex.split(output)
+    match_blocks = SPLIT_REGEX.split(output)
     while '' in match_blocks:
         match_blocks.remove('')
 
-    rule_regex = re.compile(r'(\w*)\s\[(.*)\]\s([.]{0,2}/)(.+)')
-    rules = rule_regex.findall(output)
+    rules = RULE_REGEX.findall(output)
 
     if not len(match_blocks) == len(rules):
         raise ValueError()
     return match_blocks, rules
 
 
-def _append_match_to_result(match, resulting_matches: dict[str, dict], rule):
-    rule_name, meta_string, _, _ = rule
-    _, offset, matched_tag, matched_string = match
-    resulting_matches.setdefault(
-        rule_name, {'rule': rule_name, 'matches': True, 'strings': [], 'meta': _parse_meta_data(meta_string)}
-    )
-    resulting_matches[rule_name]['strings'].append((int(offset, 16), matched_tag, matched_string))
-
-
 def _parse_meta_data(meta_data_string: str) -> dict[str, str | bool | int]:
     '''
     Will be of form 'item0=lowercaseboolean0,item1="str1",item2=int2,...'

src/analysis/plugin/compat.py

@@ -1,7 +1,14 @@
-import yara
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 
 from statistic.analysis_stats import ANALYSIS_STATS_LIMIT
 
+if TYPE_CHECKING:
+    import yara
+    from helperFunctions.types import NewPluginKind
+
 
 class AnalysisBasePluginAdapterMixin:
     """A mixin that makes AnalysisPluginV0 compatible to AnalysisBasePlugin"""
@@ -11,39 +18,39 @@ def start(self):
         pass
 
     @property
-    def NAME(self):  # noqa: N802
+    def NAME(self: NewPluginKind):  # noqa: N802
         return self.metadata.name
 
     @property
-    def DESCRIPTION(self):  # noqa: N802
+    def DESCRIPTION(self: NewPluginKind):  # noqa: N802
         return self.metadata.description
 
     @property
-    def DEPENDENCIES(self):  # noqa: N802
+    def DEPENDENCIES(self: NewPluginKind):  # noqa: N802
         return self.metadata.dependencies
 
     @property
-    def VERSION(self):  # noqa: N802
+    def VERSION(self: NewPluginKind):  # noqa: N802
         return str(self.metadata.version)
 
     @property
-    def RECURSIVE(self):  # noqa: N802
+    def RECURSIVE(self: NewPluginKind):  # noqa: N802
         return False
 
     @property
-    def TIMEOUT(self):  # noqa: N802
+    def TIMEOUT(self: NewPluginKind):  # noqa: N802
         return self.metadata.timeout
 
     @property
-    def SYSTEM_VERSION(self):  # noqa: N802
+    def SYSTEM_VERSION(self: NewPluginKind):  # noqa: N802
         return self.metadata.system_version
 
     @property
-    def MIME_BLACKLIST(self):  # noqa: N802
+    def MIME_BLACKLIST(self: NewPluginKind):  # noqa: N802
         return self.metadata.mime_blacklist
 
     @property
-    def MIME_WHITELIST(self):  # noqa: N802
+    def MIME_WHITELIST(self: NewPluginKind):  # noqa: N802
         return self.metadata.mime_whitelist
 
     @property