feat: added D-Link DCS FW unpacker and YARA MIME scanner

Author: jstucke

fact_extractor/helperFunctions/yara.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+import yara
+
+_RULE_DIR = Path(__file__).parent.parent / 'signatures/file_types'
+_COMPILED_RULES = _RULE_DIR / 'compiled.yarc'
+_rule_files = list(_RULE_DIR.glob('*.yara'))
+_rule_creation_time = _COMPILED_RULES.stat().st_mtime if _COMPILED_RULES.exists() else 0
+try:
+    # if any rule file was changed after the compiled rule file, recompile the rules
+    if any(f.stat().st_mtime > _rule_creation_time for f in _rule_files):
+        _YARA_RULES = yara.compile(filepaths={f.stem: str(f) for f in _rule_files})
+        _YARA_RULES.save(str(_COMPILED_RULES))
+    else:
+        _YARA_RULES = yara.load(str(_COMPILED_RULES))
+except SyntaxError:
+    logging.error(f'File from {_RULE_DIR} could not be compiled with YARA')
+    _YARA_RULES = yara.compile(source='')
+
+
+def get_yara_magic(path: str | Path) -> str:
+    for match in _YARA_RULES.match(str(path)):
+        return match.meta.get('MIME', match.rule)
+    return 'application/octet-stream'

fact_extractor/plugins/unpacking/dlink/code/dlink_dcs.py

@@ -0,0 +1,206 @@
+from __future__ import annotations
+
+import struct
+from dataclasses import dataclass
+from pathlib import Path
+
+NAME = 'd-link dcs'
+MIME_PATTERNS = ['firmware/dlink-dcs', 'firmware/dlink-dcs-enc']
+VERSION = '0.1.0'
+
+# loosely based on the "D-Link Firmware unpacker V1.0" from http://www.hardwarefetish.com
+
+HEADER_MAGIC = 0xAA7EC55B
+HEADER_SIZE = 0x10
+SECTION_HEADER_MAGIC = 0xA55A
+SECTION_HEADER_SIZE = 0x40
+# fmt: off
+RANDOM_TABLES = [
+    [
+        0x07CC, 0x33A8, 0xAEA8, 0x5A00, 0xDD42, 0x518E, 0x45B2, 0x0DED, 0x6D05, 0xD447, 0x9C3F, 0x4678, 0xB113, 0x7BF8,
+        0x4510, 0xB3F8, 0x9C21, 0xFD90, 0x8055, 0xB43A, 0xCE09, 0x5294, 0x6100, 0xD7DC, 0x989A, 0x4EC7, 0xC3D3, 0xFF7E,
+        0xEF08, 0xA4DA, 0xD9D3, 0xF6D4, 0xD883, 0x887B, 0x50D4, 0xB5C5, 0xDA09, 0x9687, 0xC3B3, 0x470F, 0x6ACE, 0x5FF2,
+        0x8D87, 0x1BE1, 0xDBEA, 0xD297, 0xCFDA, 0x780B, 0xD027, 0x502F, 0x2C45, 0x9E31, 0xA2C3, 0x8D45, 0x760D, 0x3B5D,
+        0xDC0D, 0x39E1, 0x3ADB, 0xCB15, 0xDEBB, 0x14AE, 0xC1E9, 0xB73E, 0x9D2A, 0x12BD, 0x6D04, 0x7733, 0xA944, 0x30B7,
+        0xBE42, 0x1413, 0x90A9, 0x4BC9, 0x2FF4, 0x6C93, 0x1E60, 0xFFCE, 0xE49E, 0xEE88, 0x4FFE, 0x10E4, 0x8CB9, 0xF2C1,
+        0x9E29, 0x02C6, 0x2E1F, 0x7A36, 0x3CA7, 0x68FA, 0x454B, 0x1B63, 0x7DA9, 0x0734, 0xD2A1, 0x1AD3, 0x19F2, 0x3FA5,
+        0x9206, 0xC336, 0x705C, 0x5049, 0xD749, 0x0105, 0x9C12, 0x073E, 0x6D98, 0xBA73, 0x070C, 0x5237, 0xA8FB, 0x570A,
+        0x631B, 0x35B4, 0x49CC, 0x0144, 0x387A, 0x77EB, 0x7B7B, 0x7522, 0xE0E5, 0xC0C6, 0x9085, 0x5E8E, 0xC7FB, 0x6326,
+        0x7961, 0xE1ED, 0xA2CC, 0x0B68, 0xA523, 0x1328, 0x5BB1, 0x7C6D, 0x142E, 0xF7C3, 0x83AB, 0x81C6, 0xB236, 0x8AB7,
+        0xD3FD, 0x5B31, 0xE1C2, 0x3718, 0x90E5, 0x2B8E, 0x385D, 0xC960, 0xA379, 0xB3D8, 0x3E82, 0x845E, 0x749E, 0xCF07,
+        0xE2ED, 0x3C99, 0x322D, 0x5C4E, 0x1E86, 0xD4F9, 0x67B6, 0xC3AA, 0xE822, 0xC367, 0x4017, 0xFC50, 0xBB2B, 0xC3C2,
+        0x7E16, 0x6D61, 0x4E79, 0x5214, 0xC893, 0x303B, 0x892C, 0x5978, 0x5BC9, 0xC189, 0x22D8, 0xFF42, 0x7561, 0x615A,
+        0x83A1, 0xEA00, 0x3061, 0x668E, 0x2699, 0x628F, 0xC2DC, 0x4520, 0x3788, 0x2A93, 0x08CA, 0x1FAA, 0xEDFA, 0x48E1,
+        0x1BFA, 0xA925, 0x0CA3,
+    ],
+    [
+        0x9A11, 0x1687, 0x5B1C, 0xEC25, 0xDF1A, 0x8B58, 0x7551, 0x3892, 0xE721, 0x36DB, 0x5B6B, 0xE664, 0xAC3C, 0xBCC5,
+        0x6A05, 0x963C, 0xED27, 0xD093, 0xBCD6, 0x4FB6, 0x936F, 0x01F6, 0x873E, 0xBE02, 0x0AC0, 0xA6E9, 0xABFD, 0x53A1,
+        0xC2E3, 0x5522, 0x6044, 0x5CF4, 0x6BA9, 0xBB60, 0x4919, 0x4AC3, 0x46B8, 0xBE6B, 0x8356, 0x2DDA, 0xF546, 0xDEC1,
+        0x143E, 0xA182, 0x9B86, 0x7E43, 0x37BF, 0x88AD, 0x4ED6, 0xF495, 0xD863, 0xE245, 0xF68B, 0x5FA2, 0xA048, 0x014B,
+        0x068B, 0x4C45, 0x54EC, 0xC96E, 0xA167, 0xB530, 0x2663, 0x0D11, 0x7090, 0x6F7C, 0x57D4, 0xB749, 0x2DE7, 0xDB2A,
+        0xE523, 0x232D, 0xB9EB, 0xF961, 0xC4B0, 0x5572, 0x77A4, 0xFC6F, 0xDE1F, 0xC67A, 0xF104, 0xB683, 0xA8BF, 0xE78F,
+        0x1625, 0x4907, 0xE8DA, 0x1CB0, 0x954C, 0x3DC6, 0xE61E, 0x36B4, 0xF2F6, 0x0C81, 0x43C5, 0x6386, 0x7BFE, 0x9B99,
+        0x1ACF, 0xA9E5, 0x76C4, 0xFFF2, 0xCD13, 0x30AF, 0xF953, 0x91C3, 0x8621, 0x70F7, 0x8E32, 0x6441, 0x3771, 0x7F36,
+        0x1AC4, 0xE031, 0x66C5, 0x30E9, 0x2938, 0x4F9F, 0x4D99, 0xBE85, 0x8D65, 0x33B7, 0xF539, 0x805B, 0x4039, 0x38FE,
+        0xE3E1, 0xBC37, 0xD497, 0xFEB1, 0x661C, 0x4B5B, 0xFEA3, 0x332F, 0x7C0B, 0xF7F7, 0xC4F2, 0x022C, 0x68EE, 0x5324,
+        0x666D, 0xA060, 0xD25A, 0x8131, 0x8091, 0x391F, 0xB21A, 0xA9C9, 0x88BE, 0xFFB3, 0x684E, 0x1623, 0x336B, 0x5D87,
+        0x967E, 0x73A4, 0x9685, 0x7A60, 0x2FDB, 0x6B1D, 0x7911, 0x95F7, 0xB678, 0x77B4, 0xC927, 0x3283, 0x6FAB, 0x8E19,
+        0x34B0, 0xD89A, 0xE13E, 0x9B1D, 0x78FA, 0xB398, 0x1C4F, 0xF98B, 0xECB8, 0xCE69, 0xA354, 0x7576, 0xCE1D, 0x0BA3,
+        0x8B9A, 0x0188, 0x692A, 0x2218, 0x752C, 0xFFB0, 0x9C78, 0xA507, 0x6ACD, 0x1589, 0x3AFE, 0x2145, 0x8D3E, 0x0425,
+        0x53C9, 0xFCE9, 0x923F, 0x8879, 0xD583, 0x737D, 0x2396, 0x4E7D, 0x2715, 0x3FE5, 0x4808, 0x13CD, 0x0E4F, 0xEB5D,
+        0x8944,
+    ],
+    [
+        0xDC6C, 0xF700, 0x14DE, 0xDDF4, 0x602A, 0x36F6, 0x5320, 0x5FDA, 0xD36F, 0xF827, 0xCAA7, 0xE8F8, 0x3325, 0xEBED,
+        0x7636, 0x374B, 0x3FB6, 0x7320, 0xC98A, 0xC82F, 0x48A3, 0x3D07, 0xEBC5, 0x9721, 0x641C, 0x2BAB, 0xDF29, 0x77EA,
+        0x39FA, 0xCA86, 0x012E, 0x1666, 0xC186, 0x160C, 0xF45A, 0x21B1, 0x4D02, 0x477A, 0x818B, 0x2071, 0x3FA1, 0x4C33,
+        0x096A, 0x72C6, 0x3820, 0x7FA0, 0xAA11, 0x77D6, 0xF2C0, 0x739B, 0x4005, 0x3B64, 0xB0A2, 0x2BCA, 0xD285, 0x14BF,
+        0x5775, 0xB1AE, 0x8CA9, 0x916F, 0x7C35, 0x8DD7, 0xA7D5, 0x3DBB, 0xA3E3, 0x9C2F, 0x5F6C, 0xF0E5, 0xE3A9, 0xE0F8,
+        0x1157, 0x234A, 0x2D2B, 0x1AC1, 0x9611, 0x654B, 0x9A61, 0x4022, 0xDD21, 0x8D22, 0xB3BE, 0x1D26, 0xC886, 0x6460,
+        0x48F0, 0x9B0B, 0x791F, 0xA066, 0x4CB9, 0x05C8, 0x31D5, 0xC8EE, 0x939F, 0xD9AB, 0x06AA, 0x3782, 0x75DA, 0x6616,
+        0x2868, 0x5984, 0x470E, 0x39BF, 0x7CCE, 0x7439, 0x5480, 0x12DF, 0xD984, 0xEEE1, 0x5302, 0xB6A5, 0x7C03, 0x06C0,
+        0xD3CB, 0x4489, 0x6B20, 0x1CBC, 0xDF94, 0xE440, 0xBD22, 0x2C4E, 0xEA08, 0xEEF7, 0xF53C, 0x7DA8, 0xC8A2, 0xFBE6,
+        0xB52A, 0x3E7D, 0x61FD, 0xDD92, 0x9801, 0xA90B, 0x1751, 0x14CF, 0x1D45, 0x6BD1, 0x27AF, 0xF6C9, 0x5AB3, 0x7AB1,
+        0xAD6F, 0xD6B6, 0x8171, 0x813A, 0x1B40, 0xEC91, 0x9DF6, 0xFAD4, 0xD0D1, 0x5B18, 0x2722, 0xBADA, 0x4A10, 0x1C5F,
+        0x3882, 0x12B2, 0x1845, 0xEDAC, 0x512F, 0x7A42, 0xCB3F, 0xE930, 0x234E, 0xE290, 0xFE00, 0x4093, 0x4E62, 0x25AF,
+        0x375C, 0xA915, 0xA060, 0xE4CB, 0x7FCB, 0x21D1, 0x6606, 0x9B0B, 0x0E62, 0x03FC, 0x95E0, 0xDF34, 0x5F15, 0xBD02,
+        0x9A0E, 0xA925, 0xD961, 0xD290, 0xBBD7, 0xF1A7, 0xC03C, 0x0D07, 0x6BE9, 0x8B7B, 0xF637, 0x8F37, 0x6E0C, 0xF437,
+        0xCFCA, 0xBC6E, 0x19E6, 0x0727, 0x6583, 0xBA46, 0xEBF2, 0xE54E, 0xDC17, 0x51F8, 0x805A, 0xEA7A, 0x55F5, 0x163A,
+        0xC9AE, 0xB50A, 0xD33C, 0x63BC, 0x5E2F, 0xAC9E, 0x364C, 0x1A06, 0x9E45, 0xF688, 0x270D, 0x0A2E, 0x8204,
+    ]
+]
+# fmt: on
+SECTION_TYPES = {
+    0: 'AUTO',
+    1: 'JFFS2',
+    2: 'YAFFS',
+    3: 'NONE',
+}
+
+
+class RandomNumberGenerator:
+    def __init__(self):
+        self._state = [0] * 3
+
+    def _get_rand(self, index: int) -> int:
+        result = RANDOM_TABLES[index][self._state[index]]
+        self._state[index] = (self._state[index] + 1) % len(RANDOM_TABLES[index])
+        return result
+
+    def get_random(self) -> int:
+        return self._get_rand(0) ^ self._get_rand(1) ^ self._get_rand(2)
+
+
+@dataclass
+class DlinkDcsHeader:
+    filesize: int
+    checksum: int
+    payload_key: int
+    section_count: int
+    header_key: int
+
+    @classmethod
+    def from_bytes(cls, data: bytes, key: int) -> DlinkDcsHeader:
+        size, checksum, payload_key, section_count, header_key = struct.unpack('<IHHHH', data)
+        return cls(
+            size ^ _pad_key(key),
+            checksum ^ key,
+            payload_key ^ key,
+            section_count ^ key,
+            header_key ^ key,
+        )
+
+    def __str__(self):
+        return (
+            f'====== Header ======\n'
+            f'size: {self.filesize:,} bytes\n'
+            f'checksum: 0x{self.checksum:x}\n'
+            f'payload key: 0x{self.payload_key:x}\n'
+            f'section count: {self.section_count}\n'
+            f'header key: 0x{self.header_key:x}'
+        )
+
+
+@dataclass
+class SectionHeader:
+    magic: int
+    mtd: int
+    section_type: int
+    size: int
+    offset: int
+    padding: bytes
+
+    @classmethod
+    def from_bytes(cls, data: bytes) -> SectionHeader:
+        values = struct.unpack('<HBBII52s', data)
+        return cls(*values)
+
+    def __str__(self):
+        return (
+            f'mtd: 0x{self.mtd:x}\n'
+            f'section_type: 0x{self.section_type:x} ({SECTION_TYPES.get(self.section_type, "unknown")})\n'
+            f'size: {self.size:,} bytes\n'
+            f'offset: 0x{self.offset:x}'
+        )
+
+
+def unpack_function(file_path: str, tmp_dir: str) -> dict:
+    with Path(file_path).open(mode='rb') as fp:
+        encoded_magic = struct.unpack('<I', fp.read(4))[0]
+        key = _derive_key(encoded_magic)
+        header = DlinkDcsHeader.from_bytes(fp.read(12), key)
+        # the key should be stored in the last field (in the XOR encrypted file it should be 0)
+        assert header.header_key == key, 'sanity check for machine code failed'
+        payload = _decrypt_payload(fp.read(), header.payload_key ^ header.header_key)
+        dump_log = _dump_sections(payload, header, Path(tmp_dir))
+    return {'output': str(header) + dump_log}
+
+
+def _derive_key(encoded_magic) -> int:
+    key_full = encoded_magic ^ HEADER_MAGIC
+    key_lower = key_full & 0xFFFF
+    key_upper = (key_full >> 16) & 0xFFFF
+    assert key_lower == key_upper, f'sanity check for encryption key failed: {key_full:x}'
+    return key_lower
+
+
+def _pad_key(key: int) -> int:
+    # pad 16-bit key to 32 bits
+    return (key << 16) ^ key
+
+
+def _decrypt_payload(data: bytes, key: int) -> bytes:
+    if len(data) % 2 != 0:
+        # length must be even => pad with zero if odd
+        data += b'\x00'
+    rng = RandomNumberGenerator()
+    result = bytearray(len(data))
+    for i in range(0, len(data), 2):
+        decoded = rng.get_random() ^ key ^ struct.unpack_from('<H', data, i)[0]
+        struct.pack_into('<H', result, i, decoded)
+    return bytes(result)
+
+
+def _dump_sections(payload: bytes, header: DlinkDcsHeader, output_dir: Path) -> str:
+    data_offset = header.section_count * SECTION_HEADER_SIZE
+    log = ''
+    for idx in range(header.section_count):
+        section_header_offset = idx * SECTION_HEADER_SIZE
+        section_header = SectionHeader.from_bytes(
+            payload[section_header_offset : section_header_offset + SECTION_HEADER_SIZE]
+        )
+        assert section_header.magic == SECTION_HEADER_MAGIC, 'wrong section magic'
+        assert section_header.size < header.filesize
+
+        output_file = output_dir / f'section_{idx:03d}'
+        output_file.write_bytes(payload[data_offset : data_offset + section_header.size])
+        data_offset += section_header.size
+        log += f'\n\n====== Section {idx} ======\n' + str(section_header) + f'\nsaved as {output_file.name}'
+    return log
+
+
+# ----> 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/dlink/test/data/test_dcs.bin

@@ -0,0 +1,21 @@
+from pathlib import Path
+
+from test.unit.unpacker.test_unpacker import TestUnpackerBase
+
+TEST_DATA_DIR = Path(__file__).parent / 'data'
+
+
+class TestDlinkDcs(TestUnpackerBase):
+    def test_unpacker_selection_generic(self):
+        self.check_unpacker_selection('firmware/dlink-dcs', 'd-link dcs')
+        self.check_unpacker_selection('firmware/dlink-dcs-enc', 'd-link dcs')
+
+    def test_extraction(self):
+        in_file = TEST_DATA_DIR / 'test_dcs.bin'
+        files, meta_data = self.unpacker.extract_files_from_file(in_file, self.tmp_dir.name)
+        assert len(files) == 2
+        name_to_file = {p.name: p for f in files if (p := Path(f))}
+        assert 'section_000' in name_to_file
+        assert name_to_file['section_000'].read_text() == 'foobar\\ntest 1234\n'
+        assert 'output' in meta_data
+        assert 'saved as section_000' in meta_data['output']

fact_extractor/plugins/unpacking/dlink/test/test_dlink_dcs.py

@@ -0,0 +1,19 @@
+rule DLink_DCS_FW {
+    meta:
+        MIME = "firmware/dlink-dcs"
+    condition:
+        uint32(0x00) == 0xAA7EC55B
+}
+
+rule DLink_DCS_FW_ENC {
+    meta:
+        MIME = "firmware/dlink-dcs-enc"
+    condition:
+        // we are only looking for encrypted files here
+        not DLink_DCS_FW
+        // uint32(0x00) = magic_enc, uint32(0x04) = size_enc
+        // magic should be 0xAA7EC55B => XOR_key = magic_enc ^ magic
+        // size = size_enc ^ XOR_key = size_enc ^ magic_enc ^ magic
+        // and size should be equal to filesize - header_size (=16) if the file is in the DCS format
+        and uint32(0x04) ^ uint32(0x00) ^ 0xAA7EC55B == filesize - 16
+}

fact_extractor/signatures/file_types/firmware.yara

@@ -0,0 +1,7 @@
+from helperFunctions.yara import get_yara_magic
+from plugins.unpacking.dlink.test.test_dlink_dcs import TEST_DATA_DIR
+
+
+def test_get_yara_magic():
+    in_file = TEST_DATA_DIR / 'test_dcs.bin'
+    assert get_yara_magic(in_file) == 'firmware/dlink-dcs-enc'

fact_extractor/test/unit/helperFunctions/test_yara.py

@@ -14,6 +14,7 @@
 from helperFunctions import magic
 from helperFunctions.config import read_list_from_config
 from helperFunctions.plugin import import_plugins
+from helperFunctions.yara import get_yara_magic
 
 
 class UnpackBase:
@@ -53,9 +54,16 @@ def get_unpacker(self, mime_type: str):
         return self.unpacker_plugins['generic/carver']
 
     def extract_files_from_file(self, file_path: str | Path, tmp_dir) -> Tuple[List, Dict]:
-        current_unpacker = self.get_unpacker(magic.from_file(file_path, mime=True))
+        current_unpacker = self.get_unpacker(self._get_mime(file_path))
         return self._extract_files_from_file_using_specific_unpacker(str(file_path), tmp_dir, current_unpacker)
 
+    @staticmethod
+    def _get_mime(file_path: str | Path) -> str:
+        mime = magic.from_file(file_path, mime=True)
+        if mime == 'application/octet-stream':
+            mime = get_yara_magic(file_path)
+        return mime
+
     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']}"""] = (

fact_extractor/unpacker/unpackBase.py

@@ -4,3 +4,4 @@ gunicorn~=23.0.0
 pytest~=8.3.5
 pytest-cov~=6.1.1
 testresources~=2.0.1
+yara-python~=4.5.4