Merge remote-tracking branch 'origin/master' into unblob-2

fact_extractor/helperFunctions/magic.py

@@ -0,0 +1,46 @@
+"""This is a wrapper around pymagic.
+It aims to provide the same API but with the ability to load multiple magic
+files in the default api.
+"""
+
+from __future__ import annotations
+
+import os
+from os import PathLike
+
+import magic as pymagic
+
+from helperFunctions.file_system import get_src_dir
+
+# On ubuntu this is provided by the libmagic-mgc package
+_default_magic = os.getenv('MAGIC', '/usr/lib/file/magic.mgc')
+_fw_magic = f'{get_src_dir()}/bin/firmware'
+_magic_file = f'{_fw_magic}:{_default_magic}'
+
+_instances = {}
+
+
+def _get_magic_instance(**kwargs):
+    """Returns an instance of pymagic.Magic"""
+    # Dicts are not hashable but sorting and creating a tuple is a valid hash
+    key = hash(tuple(sorted(kwargs.items())))
+    instance = _instances.get(key)
+    if instance is None:
+        instance = _instances[key] = pymagic.Magic(**kwargs)
+    return instance
+
+
+def from_file(filename: bytes | str | PathLike, magic_file: str | None = _magic_file, **kwargs) -> str:
+    """Like pymagic's ``magic.from_file`` but it accepts all keyword arguments
+    that ``magic.Magic`` accepts.
+    """
+    instance = _get_magic_instance(magic_file=magic_file, **kwargs)
+    return instance.from_file(filename)
+
+
+def from_buffer(buf: bytes | str, magic_file: str | None = _magic_file, **kwargs) -> str:
+    """Like pymagic's ``magic.from_buffer`` but it accepts all keyword arguments
+    that ``magic.Magic`` accepts.
+    """
+    instance = _get_magic_instance(magic_file=magic_file, **kwargs)
+    return instance.from_buffer(buf)

fact_extractor/helperFunctions/statistics.py

@@ -1,17 +1,20 @@
-from configparser import ConfigParser
+from __future__ import annotations
+
 from contextlib import suppress
-from pathlib import Path
-from typing import Dict, List
+from typing import TYPE_CHECKING
 
 from common_helper_files import safe_rglob
-from common_helper_unpacking_classifier import (
-    avg_entropy, get_binary_size_without_padding, is_compressed
-)
-from fact_helper_file import get_file_type_from_path
+from common_helper_unpacking_classifier import avg_entropy, get_binary_size_without_padding, is_compressed
+
+from helperFunctions import magic
 from helperFunctions.config import read_list_from_config
 
+if TYPE_CHECKING:
+    from configparser import ConfigParser
+    from pathlib import Path
+
 
-def add_unpack_statistics(extraction_dir: Path, meta_data: Dict):
+def add_unpack_statistics(extraction_dir: Path, meta_data: dict):
     unpacked_files, unpacked_directories = 0, 0
     for extracted_item in safe_rglob(extraction_dir):
         if extracted_item.is_file():
@@ -23,21 +26,28 @@ def add_unpack_statistics(extraction_dir: Path, meta_data: Dict):
     meta_data['number_of_unpacked_directories'] = unpacked_directories
 
 
-def get_unpack_status(file_path: str, binary: bytes, extracted_files: List[Path], meta_data: Dict, config: ConfigParser):
+def get_unpack_status(
+    file_path: str, binary: bytes, extracted_files: list[Path], meta_data: dict, config: ConfigParser
+):
     meta_data['summary'] = []
     meta_data['entropy'] = avg_entropy(binary)
 
     if not extracted_files and meta_data.get('number_of_excluded_files', 0) == 0:
-        if get_file_type_from_path(file_path)['mime'] in read_list_from_config(config, 'ExpertSettings', 'compressed_file_types')\
-                or not is_compressed(binary, compress_entropy_threshold=config.getfloat('ExpertSettings', 'unpack_threshold'), classifier=avg_entropy):
+        if magic.from_file(file_path, mime=True) in read_list_from_config(
+            config, 'ExpertSettings', 'compressed_file_types'
+        ) or not is_compressed(
+            binary,
+            compress_entropy_threshold=config.getfloat('ExpertSettings', 'unpack_threshold'),
+            classifier=avg_entropy,
+        ):
             meta_data['summary'] = ['unpacked']
         else:
             meta_data['summary'] = ['packed']
     else:
         _detect_unpack_loss(binary, extracted_files, meta_data, config.getint('ExpertSettings', 'header_overhead'))
 
 
-def _detect_unpack_loss(binary: bytes, extracted_files: List[Path], meta_data: Dict, header_overhead: int):
+def _detect_unpack_loss(binary: bytes, extracted_files: list[Path], meta_data: dict, header_overhead: int):
     decoding_overhead = 1 - meta_data.get('encoding_overhead', 0)
     cleaned_size = get_binary_size_without_padding(binary) * decoding_overhead - header_overhead
     size_of_extracted_files = _total_size_of_extracted_files(extracted_files)
@@ -46,7 +56,7 @@ def _detect_unpack_loss(binary: bytes, extracted_files: List[Path], meta_data: D
     meta_data['summary'] = ['data lost'] if cleaned_size > size_of_extracted_files else ['no data lost']
 
 
-def _total_size_of_extracted_files(extracted_files: List[Path]) -> int:
+def _total_size_of_extracted_files(extracted_files: list[Path]) -> int:
     total_size = 0
     for item in extracted_files:
         with suppress(OSError):

fact_extractor/install/common.py

@@ -1,11 +1,14 @@
 import logging
-import os
-from contextlib import suppress
+import subprocess as sp
 from pathlib import Path
 
 from helperFunctions.config import load_config
 from helperFunctions.install import (
-    apt_install_packages, apt_update_sources, pip_install_packages, load_requirements_file
+    OperateInDirectory,
+    apt_install_packages,
+    apt_update_sources,
+    load_requirements_file,
+    pip_install_packages,
 )
 
 APT_DEPENDENCIES = {
@@ -30,13 +33,35 @@
     ],
 }
 PIP_DEPENDENCY_FILE = Path(__file__).parent.parent.parent / 'requirements-common.txt'
+BIN_DIR = Path(__file__).parent.parent / 'bin'
 
 
 def install_apt_dependencies(distribution: str):
     apt_install_packages(*APT_DEPENDENCIES['common'])
     apt_install_packages(*APT_DEPENDENCIES[distribution])
 
 
+def _install_magic():
+    with OperateInDirectory(BIN_DIR):
+        sp.run(
+            [
+                'wget',
+                '--output-document',
+                'firmware.xz',
+                'https://github.com/fkie-cad/firmware-magic-database/releases/download/v0.2.2/firmware.xz',
+            ],
+            check=True,
+        )
+        sp.run(
+            [
+                'unxz',
+                '--force',
+                'firmware.xz',
+            ],
+            check=False,
+        )
+
+
 def main(distribution):
     logging.info('Updating package lists')
     apt_update_sources()
@@ -45,13 +70,13 @@ def main(distribution):
     install_apt_dependencies(distribution)
     pip_install_packages(*load_requirements_file(PIP_DEPENDENCY_FILE))
 
-    # make bin dir
-    with suppress(FileExistsError):
-        os.mkdir('../bin')
+    BIN_DIR.mkdir(exist_ok=True)
+
+    _install_magic()
 
     config = load_config('main.cfg')
     data_folder = config.get('unpack', 'data_folder')
-    os.makedirs(str(Path(data_folder, 'files')), exist_ok=True)
-    os.makedirs(str(Path(data_folder, 'reports')), exist_ok=True)
+    Path(data_folder, 'files').mkdir(parents=True, exist_ok=True)
+    Path(data_folder, 'reports').mkdir(exist_ok=True)
 
     return 0

fact_extractor/install/pre_install.sh

@@ -4,7 +4,7 @@ echo "Install Pre-Install Requirements"
 (apt-get update && apt-get install sudo) || true
 
 sudo apt-get update
-sudo apt-get -y install git apt-transport-https ca-certificates curl software-properties-common wget libmagic-dev
+sudo apt-get -y install git apt-transport-https ca-certificates curl software-properties-common wget libmagic-dev xz-utils
 
 IS_VENV=$(python3 -c 'import sys; print(sys.exec_prefix!=sys.base_prefix)')
 if [[ $IS_VENV == "False" ]]

fact_extractor/plugins/unpacking/generic_fs/code/generic_fs.py

@@ -1,18 +1,27 @@
-'''
+"""
 This plugin mounts filesystem images and extracts their content
-'''
+"""
+
 import re
 from shlex import split
-from subprocess import run, PIPE, STDOUT
+from subprocess import PIPE, STDOUT, run
 from tempfile import TemporaryDirectory
 from time import sleep
 
-from fact_helper_file import get_file_type_from_path
+from helperFunctions import magic
 
 NAME = 'genericFS'
 MIME_PATTERNS = [
-    'filesystem/btrfs', 'filesystem/dosmbr', 'filesystem/f2fs', 'filesystem/jfs', 'filesystem/minix',
-    'filesystem/reiserfs', 'filesystem/romfs', 'filesystem/udf', 'filesystem/xfs', 'generic/fs',
+    'filesystem/btrfs',
+    'filesystem/dosmbr',
+    'filesystem/f2fs',
+    'filesystem/jfs',
+    'filesystem/minix',
+    'filesystem/reiserfs',
+    'filesystem/romfs',
+    'filesystem/udf',
+    'filesystem/xfs',
+    'generic/fs',
 ]
 VERSION = '0.6.1'
 TYPES = {
@@ -28,7 +37,7 @@
 
 
 def unpack_function(file_path, tmp_dir):
-    mime_type = get_file_type_from_path(file_path)['mime']
+    mime_type = magic.from_file(file_path, mime=True)
     if mime_type == 'filesystem/dosmbr':
         output = _mount_from_boot_record(file_path, tmp_dir)
     else:

fact_extractor/test/unit/test_mime.py

@@ -0,0 +1,14 @@
+from pathlib import Path
+
+from helperFunctions import magic
+from helperFunctions.file_system import get_fact_bin_dir, get_test_data_dir
+
+
+def test_magic():
+    firmware_magic_path = Path(get_fact_bin_dir()) / 'firmware'
+    assert firmware_magic_path.is_file()
+
+    assert (
+        magic.from_file(f'{get_test_data_dir()}/ros_header', mime=True) == 'firmware/ros'
+    ), 'firmware-magic-database is not loaded'
+    assert magic.from_file(f'{get_test_data_dir()}/container/test.zip', mime=True) == 'application/zip'

fact_extractor/unpacker/unpackBase.py

@@ -1,20 +1,24 @@
+# noqa: N999
+from __future__ import annotations
+
+import fnmatch
 import logging
 from os import getgid, getuid
 from subprocess import PIPE, Popen
 from time import time
-import fnmatch
 from typing import Callable, Dict, List, Tuple
 
 from common_helper_files import get_files_in_dir
-from fact_helper_file import get_file_type_from_path
+
+from helperFunctions import magic
 from helperFunctions.config import read_list_from_config
 from helperFunctions.plugin import import_plugins
 
 
-class UnpackBase(object):
-    '''
+class UnpackBase:
+    """
     The unpacker module unpacks all files included in a file
-    '''
+    """
 
     def __init__(self, config=None, extract_everything: bool = False):
         self.config = config
@@ -36,7 +40,7 @@ def load_plugins(self):
 
     def _set_whitelist(self):
         self.blacklist = read_list_from_config(self.config, 'unpack', 'blacklist')
-        logging.debug(f'''Ignore (Blacklist): {', '.join(self.blacklist)}''')
+        logging.debug(f"""Ignore (Blacklist): {', '.join(self.blacklist)}""")
         for item in self.blacklist:
             self.register_plugin(item, self.unpacker_plugins['generic/nop'])
 
@@ -46,29 +50,33 @@ def register_plugin(self, mime_type: str, unpacker_name_and_function: Tuple[Call
     def get_unpacker(self, mime_type: str):
         if mime_type in list(self.unpacker_plugins.keys()):
             return self.unpacker_plugins[mime_type]
-        else:
-            return self.unpacker_plugins['generic/carver']
+        return self.unpacker_plugins['generic/carver']
 
     def extract_files_from_file(self, file_path: str, tmp_dir) -> Tuple[List, Dict]:
-        current_unpacker = self.get_unpacker(get_file_type_from_path(file_path)['mime'])
+        current_unpacker = self.get_unpacker(magic.from_file(file_path, mime=True))
         return self._extract_files_from_file_using_specific_unpacker(file_path, tmp_dir, current_unpacker)
 
     def unpacking_fallback(self, file_path, tmp_dir, old_meta, fallback_plugin_mime) -> Tuple[List, Dict]:
         fallback_plugin = self.unpacker_plugins[fallback_plugin_mime]
-        old_meta[f'''0_FALLBACK_{old_meta['plugin_used']}'''] = f'''{old_meta['plugin_used']} (failed) -> {fallback_plugin_mime} (fallback)'''
-        if 'output' in old_meta.keys():
-            old_meta[f'''0_ERROR_{old_meta['plugin_used']}'''] = old_meta['output']
-        return self._extract_files_from_file_using_specific_unpacker(file_path, tmp_dir, fallback_plugin, meta_data=old_meta)
+        old_meta[f"""0_FALLBACK_{old_meta['plugin_used']}"""] = (
+            f"""{old_meta['plugin_used']} (failed) -> {fallback_plugin_mime} (fallback)"""
+        )
+        if 'output' in old_meta:
+            old_meta[f"""0_ERROR_{old_meta['plugin_used']}"""] = old_meta['output']
+        return self._extract_files_from_file_using_specific_unpacker(
+            file_path, tmp_dir, fallback_plugin, meta_data=old_meta
+        )
 
     def _should_ignore(self, file):
         path = str(file)
-        for pattern in self.exclude:
-            if fnmatch.fnmatchcase(path, pattern):
-                return True
-        return False
+        return any(fnmatch.fnmatchcase(path, pattern) for pattern in self.exclude)
 
-    def _extract_files_from_file_using_specific_unpacker(self, file_path: str, tmp_dir: str, selected_unpacker, meta_data: dict = None) -> Tuple[List, Dict]:
-        unpack_function, name, version = selected_unpacker  # TODO Refactor register method to directly use four parameters instead of three
+    def _extract_files_from_file_using_specific_unpacker(
+        self, file_path: str, tmp_dir: str, selected_unpacker, meta_data: dict | None = None
+    ) -> Tuple[List, Dict]:
+        unpack_function, name, version = (
+            selected_unpacker  # TODO Refactor register method to directly use four parameters instead of three
+        )
 
         if meta_data is None:
             meta_data = {}
@@ -81,7 +89,7 @@ def _extract_files_from_file_using_specific_unpacker(self, file_path: str, tmp_d
             additional_meta = unpack_function(file_path, tmp_dir)
         except Exception as error:
             logging.debug(f'Unpacking of {file_path} failed: {error}', exc_info=True)
-            additional_meta = {'error': f'{type(error)}: {str(error)}'}
+            additional_meta = {'error': f'{type(error)}: {error!s}'}
         if isinstance(additional_meta, dict):
             meta_data.update(additional_meta)
 
@@ -101,7 +109,7 @@ def _extract_files_from_file_using_specific_unpacker(self, file_path: str, tmp_d
         meta_data['number_of_excluded_files'] = excluded_count
         return out, meta_data
 
-    def change_owner_back_to_me(self, directory: str = None, permissions: str = 'u+r'):
+    def change_owner_back_to_me(self, directory: str, permissions: str = 'u+r'):
         with Popen(f'sudo chown -R {getuid()}:{getgid()} {directory}', shell=True, stdout=PIPE, stderr=PIPE) as pl:
             pl.communicate()
         self.grant_read_permission(directory, permissions)

requirements-unpackers.txt

@@ -1,7 +1,7 @@
 # FixMe: deprecated
 pluginbase~=1.0.1
 git+https://github.com/fkie-cad/common_helper_unpacking_classifier.git
-git+https://github.com/fkie-cad/fact_helper_file.git
+python-magic
 patool~=2.2.0
 # jffs2: jefferson + deps
 git+https://github.com/sviehb/[email protected]