Retrieve FDT when present

Author: AT0myks

pyproject.toml

@@ -31,6 +31,7 @@ dependencies = [
     "pakler ~= 0.2.0",
     "pybcl ~= 1.0.0",
     "pycramfs ~= 1.1.0",
+    "pyfdt >= 0.3",
     "PySquashfsImage ~= 0.9.0",
     "python-lzo ~= 1.15",
     "ubi-reader != 0.8.7, != 0.8.8"

reolinkfw/__init__.py

@@ -6,6 +6,7 @@
 import posixpath
 import re
 import zlib
+from ast import literal_eval
 from collections.abc import Iterator, Mapping
 from contextlib import redirect_stdout
 from ctypes import sizeof
@@ -23,13 +24,15 @@
 from pakler import PAK, Section, is_pak_file
 from pycramfs import Cramfs
 from pycramfs.extract import extract_dir as extract_cramfs
+from pyfdt.pyfdt import Fdt, FdtBlobParse
 from PySquashfsImage import SquashFsImage
 from PySquashfsImage.extract import extract_dir as extract_squashfs
 from ubireader.ubifs import ubifs as ubifs_
 from ubireader.ubifs.output import extract_files as extract_ubifs
 from ubireader.ubi_io import ubi_file
 from ubireader.utils import guess_leb_size
 
+from reolinkfw.fdt import RE_FDT_HEADER, FDTHeader
 from reolinkfw.tmpfile import TempFile
 from reolinkfw.typedefs import Buffer, Files, StrPath, StrPathURL
 from reolinkfw.ubifs import UBIFS
@@ -73,6 +76,8 @@
 RE_MSTAR = re.compile(FileType.UIMAGE.value + b".{24}\x11.\x02.{33}", re.DOTALL)
 RE_UBOOT = re.compile(b"U-Boot [0-9]{4}\.[0-9]{2}.*? \(.+?\)")
 
+DUMMY = object()
+
 
 class ReolinkFirmware(PAK):
 
@@ -84,6 +89,7 @@ def __init__(self, fd: BinaryIO, offset: int = 0, closefd: bool = True) -> None:
         self._kernel_section_name = self._get_kernel_section_name()
         self._kernel_section = None
         self._kernel = None
+        self._fdt = DUMMY
         self._sdict = {s.name: s for s in self}
         self._open_files = 1
         self._fs_sections = [s for s in self if s.name in FS_SECTIONS]
@@ -141,6 +147,19 @@ def kernel(self) -> bytes:
         self._kernel = self._decompress_kernel()
         return self._kernel
 
+    @property
+    def fdt(self) -> Optional[Fdt]:
+        if self._fdt is not DUMMY:
+            return self._fdt  # type: ignore
+        self._fdt = self._get_fdt()
+        return self._fdt
+
+    @property
+    def fdt_json(self) -> Optional[dict[str, Any]]:
+        if self.fdt is not None:
+            return literal_eval(self.fdt.to_json().replace("null", "None"))
+        return None
+
     def _fdclose(self, fd: BinaryIO) -> None:
         self._open_files -= 1
         if self._closefd and not self._open_files:
@@ -200,6 +219,32 @@ def _decompress_kernel(self) -> bytes:
             return zlib.decompress(data[start:], wbits=31)
         raise Exception("unreachable")
 
+    def _get_fdt(self) -> Optional[Fdt]:
+        # At most 2 FDTs can be found in a firmware, and usually only one.
+        # Most of the time it's in the fdt section or in the decompressed kernel.
+        # Hardware versions starting with IPC_30 or IPC_32 have 1 FDT
+        # in the decompressed kernel.
+        # Hardware versions starting with IPC_35, IPC_36 or IPC_38 have no FDT.
+        # HI3536CV100 -> only firmwares with FDT in kernel_section
+        # Some firmwares with one FDT in the fdt section have a 2nd FDT
+        # in the U-Boot section with no model.
+        match = data = None
+        if "fdt" in self._sdict:
+            # Reolink Duo 1: 2 FDTs, section starts with header FKLR (Reolink FDT?)
+            # Some NVRs: 2 FDTs
+            data = self.extract_section(self["fdt"])
+            match = RE_FDT_HEADER.search(data)
+        elif (match := RE_FDT_HEADER.search(self.kernel_section)) is not None:
+            data = self.kernel_section
+        elif (match := RE_FDT_HEADER.search(self.kernel)) is not None:
+            data = self.kernel
+        if match is not None and data is not None:
+            start = match.start()
+            hdr = FDTHeader.from_buffer_copy(data, start)
+            end = start + hdr.totalsize
+            return FdtBlobParse(io.BytesIO(data[start:end])).to_fdt()
+        return None
+
     def open(self, section: Section) -> SectionFile:
         self._open_files += 1
         return SectionFile(self._fd, section, self._fdclose)
@@ -338,6 +383,9 @@ def extract(self, dest: Optional[Path] = None, force: bool = False) -> None:
         if (kcfg := self.get_kernel_config()) is not None:
             with open(dest / ".config", mode) as f:
                 f.write(kcfg)
+        if self.fdt is not None:
+            with open(dest / "camera.dts", 'w' if force else 'x', encoding="utf8") as f:
+                f.write(self.fdt.to_dts())
 
 
 async def download(url: StrOrURL) -> Union[bytes, int]:

reolinkfw/fdt.py

@@ -0,0 +1,31 @@
+import re
+from ctypes import BigEndianStructure, c_uint32
+
+# Regex for a fdt_header with version=17 and last_comp_version=16.
+RE_FDT_HEADER = re.compile(b"\xD0\x0D\xFE\xED.{16}\x00{3}\x11\x00{3}\x10.{12}", re.DOTALL)
+
+
+class FDTHeader(BigEndianStructure):
+    _fields_ = [
+        ("magic", c_uint32),
+        ("totalsize", c_uint32),
+        ("off_dt_struct", c_uint32),
+        ("off_dt_strings", c_uint32),
+        ("off_mem_rsvmap", c_uint32),
+        ("version", c_uint32),
+        ("last_comp_version", c_uint32),
+        ("boot_cpuid_phys", c_uint32),
+        ("size_dt_strings", c_uint32),
+        ("size_dt_struct", c_uint32),
+    ]
+
+    magic: int
+    totalsize: int
+    off_dt_struct: int
+    off_dt_strings: int
+    off_mem_rsvmap: int
+    version: int
+    last_comp_version: int
+    boot_cpuid_phys: int
+    size_dt_strings: int
+    size_dt_struct: int