check, fix, and format code with new ruff version

Author: jstucke

src/analysis/YaraPluginBase.py

@@ -105,12 +105,12 @@ def _append_match_to_result(match, resulting_matches: dict[str, dict], rule):
 
 
 def _parse_meta_data(meta_data_string: str) -> dict[str, str | bool | int]:
-    '''
+    """
     Will be of form 'item0=lowercaseboolean0,item1="str1",item2=int2,...'
-    '''
+    """
     try:
         # YARA insert backslashes before single quotes in the meta output and the YAML parser doesn't like that
-        meta_data_string = meta_data_string.replace(r"\'", "'")
+        meta_data_string = meta_data_string.replace(r'\'', "'")
         meta_data = yaml.safe_load(f'{{{meta_data_string.replace("=", ": ")}}}')
         assert isinstance(meta_data, dict)
         return meta_data

src/helperFunctions/install.py

@@ -53,7 +53,7 @@ def remove_folder(folder_name: str):
         shutil.rmtree(folder_name)
     except PermissionError:
         logging.debug(f'Falling back on root permission for deleting {folder_name}')
-        subprocess.run(f'sudo rm -rf {folder_name}', shell=True)
+        subprocess.run(f'sudo rm -rf {folder_name}', shell=True, check=False)
     except Exception as exception:
         raise InstallationError(exception) from None
 
@@ -70,7 +70,7 @@ def log_current_packages(packages: tuple[str], install: bool = True):
 
 
 def _run_shell_command_raise_on_return_code(command: str, error: str, add_output_on_error=False) -> str:
-    cmd_process = subprocess.run(command, shell=True, stdout=PIPE, stderr=STDOUT, text=True)
+    cmd_process = subprocess.run(command, shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False)
     if cmd_process.returncode != 0:
         if add_output_on_error:
             error = f'{error}\n{cmd_process.stdout}'
@@ -151,7 +151,9 @@ def check_if_command_in_path(command: str) -> bool:
 
     :param command: Command to check.
     """
-    command_process = subprocess.run(f'command -v {command}', shell=True, stdout=DEVNULL, stderr=DEVNULL, text=True)
+    command_process = subprocess.run(
+        f'command -v {command}', shell=True, stdout=DEVNULL, stderr=DEVNULL, text=True, check=False
+    )
     return command_process.returncode == 0
 
 
@@ -178,7 +180,7 @@ def install_github_project(project_path: str, commands: list[str]):
     with OperateInDirectory(folder_name, remove=True):
         error = None
         for command in commands:
-            cmd_process = subprocess.run(command, shell=True, stdout=PIPE, stderr=STDOUT, text=True)
+            cmd_process = subprocess.run(command, shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False)
             if cmd_process.returncode != 0:
                 error = f'Error while processing github project {project_path}!\n{cmd_process.stdout}'
                 break
@@ -189,7 +191,9 @@ def install_github_project(project_path: str, commands: list[str]):
 
 def _checkout_github_project(github_path: str, folder_name: str):
     clone_url = f'https://www.github.com/{github_path}'
-    git_process = subprocess.run(f'git clone {clone_url}', shell=True, stdout=DEVNULL, stderr=DEVNULL, text=True)
+    git_process = subprocess.run(
+        f'git clone {clone_url}', shell=True, stdout=DEVNULL, stderr=DEVNULL, text=True, check=False
+    )
     if git_process.returncode != 0:
         raise InstallationError(f'Cloning from github failed for project {github_path}\n {clone_url}')
     if not Path('.', folder_name).exists():

src/helperFunctions/yara_binary_search.py

@@ -38,7 +38,7 @@ def _execute_yara_search(self, rule_file_path: str, target_path: str | None = No
         """
         compiled_flag = '-C' if Path(rule_file_path).read_bytes().startswith(b'YARA') else ''
         command = f'yara -r {compiled_flag} {rule_file_path} {target_path or self.db_path}'
-        yara_process = subprocess.run(command, shell=True, stdout=PIPE, stderr=STDOUT, text=True)
+        yara_process = subprocess.run(command, shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False)
         return yara_process.stdout
 
     def _execute_yara_search_for_single_firmware(self, rule_file_path: str, firmware_uid: str) -> str:

src/install.py

@@ -149,7 +149,9 @@ def install_statistic_cronjob():
     crontab_file_path = current_dir.parent / 'update_statistic.cron'
     cron_content = f'0    *    *    *    *    {statistic_update_script_path} > /dev/null 2>&1\n'
     crontab_file_path.write_text(cron_content)
-    crontab_process = subprocess.run(f'crontab {crontab_file_path}', shell=True, stdout=PIPE, stderr=STDOUT, text=True)
+    crontab_process = subprocess.run(
+        f'crontab {crontab_file_path}', shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False
+    )
     if crontab_process.returncode != 0:
         logging.error(crontab_process.stdout)
     else:

src/install/backend.py

@@ -55,6 +55,7 @@ def main(skip_docker, distribution):
         shell=True,
         capture_output=True,
         text=True,
+        check=False,
     )
     if yarac_process.returncode != 0:
         raise InstallationError('Failed to compile yara test signatures')
@@ -72,7 +73,7 @@ def _install_docker_images():
     logging.info('Pulling fact extraction container')
 
     docker_process = subprocess.run(
-        'docker pull fkiecad/fact_extractor', shell=True, stdout=PIPE, stderr=STDOUT, text=True
+        'docker pull fkiecad/fact_extractor', shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False
     )
     if docker_process.returncode != 0:
         raise InstallationError(f'Failed to pull extraction container:\n{docker_process.stdout}')
@@ -89,10 +90,15 @@ def _create_firmware_directory():
 
     data_dir_name = config.backend.firmware_file_storage_directory
     mkdir_process = subprocess.run(
-        f'sudo mkdir -p --mode=0744 {data_dir_name}', shell=True, stdout=PIPE, stderr=STDOUT, text=True
+        f'sudo mkdir -p --mode=0744 {data_dir_name}', shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False
     )
     chown_process = subprocess.run(
-        f'sudo chown {os.getuid()}:{os.getgid()} {data_dir_name}', shell=True, stdout=PIPE, stderr=STDOUT, text=True
+        f'sudo chown {os.getuid()}:{os.getgid()} {data_dir_name}',
+        shell=True,
+        stdout=PIPE,
+        stderr=STDOUT,
+        text=True,
+        check=False,
     )
     if not all(code == 0 for code in (mkdir_process.returncode, chown_process.returncode)):
         raise InstallationError(
@@ -122,26 +128,28 @@ def _install_plugins(distribution, skip_docker, only_docker=False):
 def _install_yara():
     yara_version = 'v4.2.3'  # must be the same version as `yara-python` in `install/requirements_common.txt`
 
-    yara_process = subprocess.run('yara --version', shell=True, stdout=PIPE, stderr=STDOUT, text=True)
+    yara_process = subprocess.run('yara --version', shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False)
     if yara_process.returncode == 0 and yara_process.stdout.strip() == yara_version.strip('v'):
         logging.info('Skipping yara installation: Already installed and up to date')
         return
 
     logging.info(f'Installing yara {yara_version}')
     archive = f'{yara_version}.zip'
     download_url = f'https://github.com/VirusTotal/yara/archive/refs/tags/{archive}'
-    wget_process = subprocess.run(f'wget {download_url}', shell=True, stdout=PIPE, stderr=STDOUT, text=True)
+    wget_process = subprocess.run(
+        f'wget {download_url}', shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False
+    )
     if wget_process.returncode != 0:
         raise InstallationError(f'Error on yara download.\n{wget_process.stdout}')
-    unzip_process = subprocess.run(f'unzip {archive}', shell=True, stdout=PIPE, stderr=STDOUT, text=True)
+    unzip_process = subprocess.run(f'unzip {archive}', shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False)
     Path(archive).unlink()
     if unzip_process.returncode != 0:
         raise InstallationError(f'Error on yara extraction.\n{unzip_process.stdout}')
-    yara_folder = [p for p in Path('.').iterdir() if p.name.startswith('yara-')][0]
+    yara_folder = [p for p in Path().iterdir() if p.name.startswith('yara-')][0]
     with OperateInDirectory(yara_folder.name, remove=True):
         os.chmod('bootstrap.sh', 0o775)  # noqa: PTH101
         for command in ['./bootstrap.sh', './configure --enable-magic', 'make -j$(nproc)', 'sudo make install']:
-            cmd_process = subprocess.run(command, shell=True, stdout=PIPE, stderr=STDOUT, text=True)
+            cmd_process = subprocess.run(command, shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False)
             if cmd_process.returncode != 0:
                 raise InstallationError(f'Error in yara installation.\n{cmd_process.stdout}')
 
@@ -152,7 +160,7 @@ def _install_checksec():
     logging.info('Installing checksec.sh')
     checksec_url = 'https://raw.githubusercontent.com/slimm609/checksec.sh/2.5.0/checksec'
     wget_process = subprocess.run(
-        f'wget -P {BIN_DIR} {checksec_url}', shell=True, stdout=PIPE, stderr=STDOUT, text=True
+        f'wget -P {BIN_DIR} {checksec_url}', shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False
     )
     if wget_process.returncode != 0:
         raise InstallationError(f'Error during installation of checksec.sh\n{wget_process.stdout}')

src/install/frontend.py

@@ -27,7 +27,7 @@
 def execute_commands_and_raise_on_return_code(commands, error=None):
     for command in commands:
         bad_return = error if error else f'execute {command}'
-        cmd_process = subprocess.run(command, shell=True, stdout=PIPE, stderr=STDOUT, text=True)
+        cmd_process = subprocess.run(command, shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False)
         if cmd_process.returncode != 0:
             raise InstallationError(f'Failed to {bad_return}\n{cmd_process.stdout}')
 
@@ -40,10 +40,15 @@ def _create_directory_for_authentication():
     factauthdir = '/'.join(dburi.split('/')[:-1])[10:]  # FIXME this should be beautified with pathlib
 
     mkdir_process = subprocess.run(
-        f'sudo mkdir -p --mode=0744 {factauthdir}', shell=True, stdout=PIPE, stderr=STDOUT, text=True
+        f'sudo mkdir -p --mode=0744 {factauthdir}', shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False
     )
     chown_process = subprocess.run(
-        f'sudo chown {os.getuid()}:{os.getgid()} {factauthdir}', shell=True, stdout=PIPE, stderr=STDOUT, text=True
+        f'sudo chown {os.getuid()}:{os.getgid()} {factauthdir}',
+        shell=True,
+        stdout=PIPE,
+        stderr=STDOUT,
+        text=True,
+        check=False,
     )
 
     if not all(return_code == 0 for return_code in [mkdir_process.returncode, chown_process.returncode]):
@@ -70,7 +75,7 @@ def _install_nginx(distribution):
             ],
             error='restore selinux context',
         )
-    nginx_process = subprocess.run('sudo nginx -s reload', shell=True, capture_output=True, text=True)
+    nginx_process = subprocess.run('sudo nginx -s reload', shell=True, capture_output=True, text=True, check=False)
     if nginx_process.returncode != 0:
         raise InstallationError(f'Failed to start nginx\n{nginx_process.stderr}')
 
@@ -109,15 +114,15 @@ def _install_docker_images(radare):
 
         with OperateInDirectory('radare'):
             docker_compose_process = subprocess.run(
-                'docker compose build', shell=True, stdout=PIPE, stderr=STDOUT, text=True
+                'docker compose build', shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False
             )
             if docker_compose_process.returncode != 0:
                 raise InstallationError(f'Failed to initialize radare container:\n{docker_compose_process.stdout}')
 
     # pull pdf report container
     logging.info('Pulling pdf report container')
     docker_process = subprocess.run(
-        'docker pull fkiecad/fact_pdf_report', shell=True, stdout=PIPE, stderr=STDOUT, text=True
+        'docker pull fkiecad/fact_pdf_report', shell=True, stdout=PIPE, stderr=STDOUT, text=True, check=False
     )
     if docker_process.returncode != 0:
         raise InstallationError(f'Failed to pull pdf report container:\n{docker_process.stdout}')

src/plugins/analysis/architecture_detection/internal/dt.py

@@ -58,7 +58,7 @@ def _get_compatible_entry(dts: str) -> str | None:
                 break
 
         cpu = cpus[cpu_name]
-        if 'compatible' not in cpu.keys():
+        if 'compatible' not in cpu:
             continue
 
         compatible = cpu['compatible']

src/plugins/analysis/binwalk/code/binwalk.py

@@ -29,6 +29,7 @@ def process_object(self, file_object):
                 stdout=PIPE,
                 stderr=STDOUT,
                 text=True,
+                check=False,
             )
             signature_analysis_result = cmd_process.stdout
             try:

src/plugins/analysis/checksec/code/checksec.py

@@ -39,7 +39,12 @@ def process_object(self, file_object):
 
 def execute_checksec_script(file_path):
     checksec_process = subprocess.run(
-        f'{SHELL_SCRIPT} --file={file_path} --format=json --extended', shell=True, stdout=PIPE, stderr=STDOUT, text=True
+        f'{SHELL_SCRIPT} --file={file_path} --format=json --extended',
+        shell=True,
+        stdout=PIPE,
+        stderr=STDOUT,
+        text=True,
+        check=False,
     )
     if checksec_process.returncode != 0:
         raise ValueError(f'Checksec script exited with non-zero return code {checksec_process.returncode}')

src/plugins/analysis/cve_lookup/code/cve_lookup.py

@@ -41,7 +41,7 @@ def process_object(self, file_object: FileObject) -> FileObject:
         cves = {'cve_results': {}}
         connection = DbConnection(f'sqlite:///{DB_PATH}')
         lookup = Lookup(file_object, connection)
-        for _key, value in file_object.processed_analysis['software_components']['result'].items():
+        for value in file_object.processed_analysis['software_components']['result'].values():
             product = value['meta']['software_name']
             version = value['meta']['version'][0]
             if product and version:

src/plugins/analysis/device_tree/internal/device_tree_utils.py

@@ -80,7 +80,7 @@ def header_has_illegal_values(header: DeviceTreeHeader, max_size: int) -> bool:
 
 
 def convert_device_tree_to_str(file_path: str | Path) -> str | None:
-    process = run(f'dtc -I dtb -O dts {file_path}', shell=True, capture_output=True)
+    process = run(f'dtc -I dtb -O dts {file_path}', shell=True, capture_output=True, check=False)
     if process.returncode != 0:
         logging.warning(
             f'The Device Tree Compiler exited with non-zero return code {process.returncode}: {process.stderr}'

src/plugins/analysis/file_type/code/file_type.py

@@ -1,13 +1,19 @@
+from __future__ import annotations
+
+import typing
+
 from fact_helper_file import get_file_type_from_path
 import pydantic
 from pydantic import Field
 
 from analysis.plugin import AnalysisPluginV0
 from analysis.plugin.compat import AnalysisBasePluginAdapterMixin
 
-import io
 from typing import List
 
+if typing.TYPE_CHECKING:
+    import io
+
 
 class AnalysisPlugin(AnalysisPluginV0, AnalysisBasePluginAdapterMixin):
     class Schema(pydantic.BaseModel):

src/plugins/analysis/kernel_config/internal/checksec_check_kernel.py

@@ -47,7 +47,7 @@ def check_kernel_config(kernel_config: str) -> dict:
             fp.write(kernel_config.encode())
             fp.seek(0)
             command = f'{CHECKSEC_PATH} --kernel={fp.name} --output=json'
-            checksec_process = subprocess.run(command, shell=True, stdout=PIPE, stderr=DEVNULL, text=True)
+            checksec_process = subprocess.run(command, shell=True, stdout=PIPE, stderr=DEVNULL, text=True, check=False)
             result = json.loads(checksec_process.stdout)
             whitelist_configs(result)
             return result

src/plugins/analysis/kernel_config/internal/kernel_config_hardening_check.py

@@ -137,6 +137,7 @@ def _get_kernel_hardening_data(kernel_config: str) -> list[list[str]]:
                 stdout=PIPE,
                 stderr=STDOUT,
                 text=True,
+                check=False,
             )
             return json.loads(kconfig_process.stdout)
     except (JSONDecodeError, KeyError):
@@ -155,11 +156,11 @@ def _add_protection_info(hardening_result: list[list[str]]) -> list[HardeningChe
 
 
 def _detach_actual_value_from_result(single_result: list[str]) -> str:
-    '''
+    """
     the result may contain the actual value after a colon
     e.g. 'FAIL: not found' or 'FAIL: "y"'
     removes actual value and returns it (or empty string if missing)
-    '''
+    """
     split_result = single_result[4].split(': ')
     single_result[4] = split_result[0]
     return ': '.join(split_result[1:]).replace('"', '')

src/plugins/analysis/kernel_config/test/test_kernel_config.py

@@ -1,4 +1,3 @@
-import glob
 from pathlib import Path
 from subprocess import CompletedProcess
 
@@ -81,7 +80,7 @@ def test_extract_ko_success(self, analysis_plugin):
         assert analysis_plugin.probably_kernel_config(result)
 
     def test_process_objects_kernel_image(self, analysis_plugin):
-        for valid_image in glob.glob(str(TEST_DATA_DIR / 'synthetic/*.image')):
+        for valid_image in (TEST_DATA_DIR / 'synthetic').glob('*.image'):
             test_file = FileObject(file_path=str(valid_image))
             test_file.processed_analysis['file_type'] = {'result': {'mime': 'application/octet-stream'}}
             test_file.processed_analysis['software_components'] = {'summary': ['Linux Kernel']}
@@ -91,7 +90,7 @@ def test_process_objects_kernel_image(self, analysis_plugin):
             assert test_file.processed_analysis[analysis_plugin.NAME]['is_kernel_config']
             assert len(test_file.processed_analysis[analysis_plugin.NAME]['kernel_config']) > 0
 
-        for bad_image in glob.glob(str(TEST_DATA_DIR / 'random_invalid/*.image')):
+        for bad_image in (TEST_DATA_DIR / 'random_invalid').glob('*.image'):
             test_file = FileObject(file_path=str(bad_image))
             test_file.processed_analysis['file_type'] = {'result': {'mime': 'application/octet-stream'}}
             test_file.processed_analysis['software_components'] = {'summary': ['Linux Kernel']}
@@ -149,7 +148,7 @@ def test_try_extract_fail():
 
 
 def test_try_extract_random_fail():
-    for fp in glob.glob(str(TEST_DATA_DIR / 'random_invalid/*.image')):
+    for fp in (TEST_DATA_DIR / 'random_invalid').glob('*.image'):
         test_file = FileObject(file_path=fp)
         test_file.processed_analysis['file_type'] = {'result': {'mime': 'application/octet-stream'}}
         test_file.processed_analysis['software_components'] = {'summary': ['Linux Kernel']}

src/plugins/analysis/known_vulnerabilities/internal/rulebook.py

@@ -32,7 +32,7 @@ def __init__(self, rule, description, reliability, score, link, short_name):  #
 
     def _make_type_assertions(self, link, rule):
         for type_assertion, error_message in [
-            (int(self.reliability) in range(0, 101), 'reliability must be between 0 and 100'),
+            (int(self.reliability) in range(101), 'reliability must be between 0 and 100'),
             (self.score in ['low', 'medium', 'high'], 'score has to be one of low, medium or high'),
             (isinstance(self.description, str), 'description must be a string'),
             (

src/scheduler/analysis/plugin.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import ctypes
 import io
 import logging
@@ -14,11 +16,13 @@
 from pydantic import BaseModel, ConfigDict
 
 import config
-from analysis.plugin import AnalysisPluginV0
-from objects.file import FileObject
 from statistic.analysis_stats import ANALYSIS_STATS_LIMIT
 from storage.fsorganizer import FSOrganizer
 
+if typing.TYPE_CHECKING:
+    from analysis.plugin import AnalysisPluginV0
+    from objects.file import FileObject
+
 
 class PluginRunner:
     class Config(BaseModel):

src/scheduler/analysis/scheduler.py

@@ -610,7 +610,7 @@ def check_exceptions(self) -> bool:
 
         :return: Boolean value stating if any attached process ran into an exception
         """
-        for _, plugin in self.analysis_plugins.items():
+        for plugin in self.analysis_plugins.values():
             if isinstance(plugin, AnalysisPluginV0):
                 continue
             if plugin.check_exceptions():

src/start_fact_frontend.py

@@ -17,6 +17,8 @@
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 
+from __future__ import annotations
+
 import logging
 import pickle
 import signal