feat: added netgear chk unpacker based on unblob

fact_extractor/helperFunctions/unblob.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import structlog
+from structlog.testing import capture_logs
+
+if TYPE_CHECKING:
+    from unblob.models import Extractor
+
+# configure unblob internal logger
+structlog.configure(
+    wrapper_class=structlog.make_filtering_bound_logger(logging.DEBUG),
+)
+
+
+def extract_file(extractor: Extractor, path: Path, tmp_dir: str) -> str:
+    # unblob uses structlog for logging, but we can capture the logs with this convenient testing function
+    with capture_logs() as log_list:
+        extractor.extract(path, Path(tmp_dir))
+        return _format_logs(log_list)
+
+
+def _format_logs(logs: list[dict]) -> str:
+    output = ''
+    for entry in logs:
+        output += '\n'.join(f'{key}: {value}' for key, value in entry.items() if key not in {'_verbosity', 'log_level'})
+    return output

fact_extractor/install/common.py

@@ -27,7 +27,7 @@ def install_apt_dependencies():
     apt_install_packages(*APT_DEPENDENCIES)
 
 
-def _install_magic(version='v0.2.8'):
+def _install_magic(version='v0.2.9'):
     with OperateInDirectory(BIN_DIR):
         sp.run(
             [

fact_extractor/plugins/unpacking/netgear/code/chk.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+import struct
+from pathlib import Path
+
+from unblob.handlers.archive.netgear.chk import CHKExtractor
+
+from helperFunctions.unblob import extract_file
+
+NAME = 'netgear-chk'
+MIME_PATTERNS = ['firmware/netgear-chk']
+VERSION = '0.1.0'
+extractor = CHKExtractor()
+
+
+def unpack_function(file_path: str, tmp_dir: str) -> dict:
+    # Netgear CHK images are simple containers with a header (usually 58 bytes) that should contain a kernel and a
+    # root filesystem. Their sizes should also be contained in the header at offsets 24 and 28 respectively.
+    # A size of 0 indicates that the file is missing (which is the case for the 2nd file of most samples).
+    # Since unblob comes with the capability to unpack these containers, we simply use its `CHKExtractor`.
+    path = Path(file_path)
+    with path.open(mode='rb') as fp:
+        fp.seek(24)
+        kernel_len, rootfs_len = struct.unpack('>II', fp.read(8))
+        logs = f'kernel size: {kernel_len}\nrootfs size: {rootfs_len}\n'
+    logs += extract_file(extractor, path, tmp_dir)
+    return {'output': logs}
+
+
+# ----> Do not edit below this line <----
+def setup(unpack_tool):
+    for item in MIME_PATTERNS:
+        unpack_tool.register_plugin(item, (unpack_function, NAME, VERSION))

fact_extractor/plugins/unpacking/netgear/test/test_plugin_chk.py

@@ -0,0 +1,40 @@
+from pathlib import Path
+
+from test.unit.unpacker.test_unpacker import TestUnpackerBase
+
+TEST_DATA_DIR = Path(__file__).parent / 'data'
+
+
+class TestNetgearCHK(TestUnpackerBase):
+    def test_unpacker_selection_generic(self):
+        self.check_unpacker_selection('firmware/netgear-chk', 'netgear-chk')
+
+    def test_extraction_single_file(self):
+        in_file = TEST_DATA_DIR / 'test.chk'
+        assert in_file.is_file(), 'test file is missing'
+        files, meta_data = self.unpacker.base._extract_files_from_file_using_specific_unpacker(
+            str(in_file),
+            self.tmp_dir.name,
+            self.unpacker.base.unpacker_plugins['firmware/netgear-chk'],
+        )
+        assert len(files) == 1, 'unpacked file number incorrect'
+        file = Path(files[0])
+        contents = file.read_bytes()
+        assert len(contents) == 18, 'unpacked file size incorrect'
+        assert contents.startswith(b'foobar'), 'payload not decrypted correctly'
+        assert 'kernel size: 18' in meta_data['output']
+
+    def test_extraction_multi_file(self):
+        in_file = TEST_DATA_DIR / 'test2.chk'
+        assert in_file.is_file(), 'test file is missing'
+        files, meta_data = self.unpacker.base._extract_files_from_file_using_specific_unpacker(
+            str(in_file),
+            self.tmp_dir.name,
+            self.unpacker.base.unpacker_plugins['firmware/netgear-chk'],
+        )
+        assert len(files) == 2, 'unpacked file number incorrect'
+        file = Path(sorted(files)[1])
+        contents = file.read_bytes()
+        assert len(contents) == 4, 'unpacked file size incorrect'
+        assert contents == b'bar\n', 'payload not decrypted correctly'
+        assert 'rootfs size: 4' in meta_data['output']

fact_extractor/plugins/unpacking/trx/code/untrx.py

@@ -1,49 +1,24 @@
-from os import path
-from tempfile import NamedTemporaryFile
+from pathlib import Path
 
 from common_helper_process.fail_safe_subprocess import execute_shell_command
 
 from helperFunctions.file_system import get_fact_bin_dir
 
 NAME = 'untrx'
 MIME_PATTERNS = ['firmware/trx']
-VERSION = '0.4'
+VERSION = '0.4.1'
 
-UNPACKER_EXECUTEABLE = path.join(get_fact_bin_dir(), 'untrx')
+UNPACKER_EXECUTABLE = Path(get_fact_bin_dir()) / 'untrx'
 
 
 def unpack_function(file_path, tmp_dir):
-    """
-    file_path specifies the input file.
-    tmp_dir should be used to store the extracted files.
-    """
-    offset = _get_trx_offset(file_path)
-    if offset > 0:
-        with NamedTemporaryFile('bw') as tf:
-            _remove_non_trx_header(file_path, tf, offset)
-            output = _unpack_trx(tf.name, tmp_dir)
-    else:
-        output = _unpack_trx(file_path, tmp_dir)
+    output = _unpack_trx(file_path, tmp_dir)
 
     return {'output': output}
 
 
-def _get_trx_offset(file_path):
-    with open(file_path, 'br') as fp:
-        content = fp.read()
-        return content.find(b'HDR0')
-
-
-def _remove_non_trx_header(source_path, target_fp, offset):
-    with open(source_path, 'br') as source_fp:
-        source_fp.seek(offset)
-        content = source_fp.read()
-        target_fp.write(content)
-        target_fp.seek(0)
-
-
 def _unpack_trx(file_path, target_dir):
-    return execute_shell_command(f'fakeroot {UNPACKER_EXECUTEABLE} {file_path} {target_dir}')
+    return execute_shell_command(f'fakeroot {UNPACKER_EXECUTABLE} {file_path} {target_dir}')
 
 
 # ----> Do not edit below this line <----

fact_extractor/plugins/unpacking/trx/test/test_plugin_untrx.py

@@ -1,26 +1,19 @@
-import os
 import zipfile
+from pathlib import Path
 
 from test.unit.unpacker.test_unpacker import TestUnpackerBase
 
-TEST_DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'data')
+TEST_DATA_DIR = Path(__file__).parent / 'data'
 
 
 class TestUntrxUnpacker(TestUnpackerBase):
     def test_unpacker_selection_generic(self):
         self.check_unpacker_selection('firmware/trx', 'untrx')
 
     def test_extraction_trx(self):
-        files, _ = self.unpacker.extract_files_from_file(os.path.join(TEST_DATA_DIR, 'trx.img'), self.tmp_dir.name)
+        files, _ = self.unpacker.extract_files_from_file(TEST_DATA_DIR / 'trx.img', self.tmp_dir.name)
         assert len(files) == 1
         with zipfile.ZipFile(files[0], 'r') as extracted_file:
-            included_file_list = [os.path.basename(f) for f in extracted_file.namelist() if os.path.basename(f)]
+            included_file_list = [name for f in extracted_file.namelist() if (name := Path(f).name)]
             for file in ['test file 3_.txt', 'testfile1', 'testfile2']:
                 assert file in included_file_list
-
-    def test_extraction_trx2(self):
-        files, _ = self.unpacker.extract_files_from_file(os.path.join(TEST_DATA_DIR, 'netgear.trx'), self.tmp_dir.name)
-        assert len(files) == 2
-        included_file_list = [os.path.basename(f) for f in files if os.path.basename(f)]
-        for file in included_file_list:
-            assert file in ['squashfs-lzma-image-x_x', 'segment1']

fact_extractor/plugins/unpacking/xiaomi_hdr/code/xiaomi_hdr.py

@@ -4,21 +4,16 @@
 
 from __future__ import annotations
 
-import logging
 from pathlib import Path
 
-import structlog
-from structlog.testing import capture_logs
 from unblob.handlers.archive.xiaomi.hdr import HDRExtractor
 
+from helperFunctions.unblob import extract_file
+
 NAME = 'Xiaomi HDR'
 MIME_PATTERNS = ['firmware/xiaomi-hdr1', 'firmware/xiaomi-hdr2']
 VERSION = '0.1.0'
 
-structlog.configure(
-    wrapper_class=structlog.make_filtering_bound_logger(logging.DEBUG),
-)
-
 
 def unpack_function(file_path: str, tmp_dir: str) -> dict:
     path = Path(file_path)
@@ -29,17 +24,8 @@ def unpack_function(file_path: str, tmp_dir: str) -> dict:
     else:
         return {'output': ''}
 
-    # unblob uses structlog for logging, but we can capture the logs with this convenient testing function
-    with capture_logs() as log_list:
-        extractor.extract(path, Path(tmp_dir))
-        return {'output': _format_logs(log_list)}
-
-
-def _format_logs(logs: list[dict]) -> str:
-    output = ''
-    for entry in logs:
-        output += '\n'.join(f'{key}: {value}' for key, value in entry.items() if key not in {'_verbosity', 'log_level'})
-    return output
+    logs = extract_file(extractor, path, tmp_dir)
+    return {'output': logs}
 
 
 # ----> Do not edit below this line <----