1import binascii
2import io
3from collections.abc import Iterable
4from pathlib import Path
5from typing import Optional, cast
6
7from structlog import get_logger
8
9from unblob.file_utils import (
10 File,
11 FileSystem,
12 InvalidInputFormat,
13 iterate_file,
14 snull,
15)
16from unblob.models import (
17 Endian,
18 Extractor,
19 ExtractResult,
20 HandlerDoc,
21 HandlerType,
22 HexString,
23 StructHandler,
24 StructParser,
25 ValidChunk,
26)
27
28logger = get_logger()
29CRC_CONTENT_OFFSET = 12 # The CRC32 value is located after 12 byte in the header
30SIGNATURE_LEN = 272 # signature_header_t contains 4 bytes of size + 12 bytes for padding x3 + 0x100 is 256 in decimal
31BLOB_MAGIC = 0x000000BEBA # Blob header magic
32
33# https://lxr.openwrt.org/source/firmware-utils/src/xiaomifw.c
34C_DEFINITIONS = r"""
35 struct hdr1_header {
36 char magic[4]; /* HDR1 */
37 uint32 signature_offset;
38 uint32 crc32;
39 uint16 unused;
40 uint16 device_id; /* RA70 */
41 uint32 blob_offsets[8];
42 } hdr1_header_t;
43
44 struct hdr2_header {
45 char magic[4]; /* HDR1 */
46 uint32 signature_offset;
47 uint32 crc32;
48 uint32 unused1;
49 uint64 device_id; /* RA70 */
50 uint64 region; /* EU */
51 uint64 unused2[2];
52 uint32 blob_offsets[8];
53 } hdr2_header_t;
54
55 struct xiaomi_blob_header {
56 uint32 magic; /* 0x0000babe */
57 uint32 flash_offset;
58 uint32 size; /* Size of blob */
59 uint16 type; /* Type of blob */
60 uint16 unused;
61 char name[32]; /* Name of blob */
62 } blob_header_t;
63
64 struct xiaomi_signature_header {
65 uint32 size;
66 uint32 padding[3];
67 uint8 content[0x100];
68 } signature_header_t;
69 """
70
71
72def calculate_crc(file: File, start_offset: int, size: int) -> int:
73 digest = 0
74 for chunk in iterate_file(file, start_offset, size):
75 digest = binascii.crc32(chunk, digest)
76 return (digest ^ -1) & 0xFFFFFFFF
77
78
79def is_valid_blob_header(blob_header) -> bool:
80 if blob_header.magic == BLOB_MAGIC:
81 return False
82 if not blob_header.size:
83 return False
84 try:
85 snull(blob_header.name).decode("utf-8")
86 except UnicodeDecodeError:
87 return False
88 return True
89
90
91def is_valid_header(header) -> bool:
92 if header.signature_offset < len(header):
93 return False
94 if not header.blob_offsets[0]: # noqa: SIM103
95 return False
96 return True
97
98
99class HDRExtractor(Extractor):
100 def __init__(self, header_struct: str):
101 self.header_struct = header_struct
102 self._struct_parser = StructParser(C_DEFINITIONS)
103
104 def extract(self, inpath: Path, outdir: Path):
105 fs = FileSystem(outdir)
106 with File.from_path(inpath) as file:
107 for output_path, start_offset, size in self.parse(file):
108 fs.carve(output_path, file, start_offset, size)
109 return ExtractResult(reports=fs.problems)
110
111 def parse(self, file: File) -> Iterable[tuple[Path, int, int]]:
112 header = self._struct_parser.parse(self.header_struct, file, Endian.LITTLE)
113 for offset in cast("Iterable", header.blob_offsets):
114 if not offset:
115 break
116
117 file.seek(offset, io.SEEK_SET)
118 blob_header = self._struct_parser.parse(
119 "blob_header_t", file, Endian.LITTLE
120 )
121 logger.debug("blob_header_t", blob_header_t=blob_header, _verbosity=3)
122 if not is_valid_blob_header(blob_header):
123 raise InvalidInputFormat("Invalid HDR blob header.")
124
125 yield (
126 (
127 Path(snull(blob_header.name).decode("utf-8")),
128 # file.tell() points to right after the blob_header == start_offset
129 file.tell(),
130 blob_header.size,
131 )
132 )
133
134
135class HDRHandlerBase(StructHandler):
136 HEADER_STRUCT = "hdr1_header_t"
137
138 def calculate_chunk(self, file: File, start_offset: int) -> Optional[ValidChunk]:
139 header = self.parse_header(file, endian=Endian.LITTLE)
140
141 if not is_valid_header(header):
142 raise InvalidInputFormat("Invalid HDR header.")
143
144 end_offset = start_offset + header.signature_offset + SIGNATURE_LEN
145
146 if not self._is_crc_valid(file, header, start_offset, end_offset):
147 raise InvalidInputFormat("CRC32 does not match in HDR header.")
148
149 return ValidChunk(start_offset=start_offset, end_offset=end_offset)
150
151 def _is_crc_valid(
152 self, file: File, header, start_offset: int, end_offset: int
153 ) -> bool:
154 computed_crc = calculate_crc(
155 file,
156 start_offset=start_offset + CRC_CONTENT_OFFSET,
157 size=end_offset - start_offset + CRC_CONTENT_OFFSET,
158 )
159 return header.crc32 == computed_crc
160
161
162class HDR1Handler(HDRHandlerBase):
163 NAME = "hdr1"
164 PATTERNS = [
165 HexString(
166 """
167 // 48 44 52 31 90 32 e2 00 02 2a 5b 6a 00 00 11 00 HDR1.2...*[j....
168 // 30 00 00 00 70 02 00 00 a4 02 e0 00 00 00 00 00 0...p...........
169 48 44 52 31
170 """
171 ),
172 ]
173 C_DEFINITIONS = C_DEFINITIONS
174 HEADER_STRUCT = "hdr1_header_t"
175 EXTRACTOR = HDRExtractor("hdr1_header_t")
176
177 DOC = HandlerDoc(
178 name="Xiaomi HDR1",
179 description="Xiaomi HDR1 firmware files feature a custom header containing metadata, CRC32 checksum, and blob offsets for embedded data. These files are used in Xiaomi devices for firmware updates.",
180 handler_type=HandlerType.ARCHIVE,
181 vendor="Xiaomi",
182 references=[],
183 limitations=[],
184 )
185
186
187class HDR2Handler(HDRHandlerBase):
188 NAME = "hdr2"
189 PATTERNS = [
190 HexString(
191 """
192 // 48 44 52 32 d4 02 78 02 68 54 e8 fa 00 00 00 00 HDR2..x.hT......
193 // 52 41 37 30 00 00 00 00 45 55 00 00 00 00 00 00 RA70....EU......
194 48 44 52 32
195 """
196 ),
197 ]
198 C_DEFINITIONS = C_DEFINITIONS
199 HEADER_STRUCT = "hdr2_header_t"
200 EXTRACTOR = HDRExtractor("hdr2_header_t")
201
202 DOC = HandlerDoc(
203 name="Xiaomi HDR2",
204 description="Xiaomi HDR2 firmware files feature a custom header with metadata, CRC32 checksum, and blob offsets for embedded data. These files also include additional fields for device ID and region information.",
205 handler_type=HandlerType.ARCHIVE,
206 vendor="Xiaomi",
207 references=[],
208 limitations=[],
209 )