Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import lzma
2from collections.abc import Iterator
3from pathlib import Path
5from unblob.file_utils import Endian, File, FileSystem, InvalidInputFormat, StructParser
6from unblob.models import (
7 Extractor,
8 ExtractResult,
9 HandlerDoc,
10 HandlerType,
11 HexString,
12 Reference,
13 StructHandler,
14 ValidChunk,
15)
17C_DEFINITIONS = r"""
18 typedef struct qca_lzma_header {
19 uint16 literal_context_bits; /* only 0x0003 observed */
20 uint16 position_bits; /* only 0x0002 observed */
21 uint32 dict_size; /* only 0x00001000 observed */
22 uint32 compressed_size;
23 uint32 decompressed_size;
24 } qca_lzma_header_t;
25"""
28class QcaLzmaExtractor(Extractor):
29 def __init__(self):
30 self._struct_parser = StructParser(C_DEFINITIONS)
32 def extract(self, inpath: Path, outdir: Path):
33 fs = FileSystem(outdir)
34 with File.from_path(inpath) as file:
35 header = self._struct_parser.parse("qca_lzma_header_t", file, Endian.LITTLE)
36 fs.write_chunks(
37 Path(f"{inpath.stem}.uncompressed"), _decompress(file, header)
38 )
39 return ExtractResult(reports=fs.problems)
42def _decompress(file: File, header, chunk_size: int = 1_024) -> Iterator[bytes]:
43 decompressor = lzma.LZMADecompressor(
44 format=lzma.FORMAT_RAW,
45 filters=[
46 {
47 "id": lzma.FILTER_LZMA1,
48 "lc": header.literal_context_bits,
49 "lp": 0,
50 "pb": header.position_bits,
51 "dict_size": header.dict_size,
52 }
53 ],
54 )
55 bytes_read = 0
57 while not decompressor.eof and bytes_read < header.compressed_size:
58 chunk = file.read(min(chunk_size, header.compressed_size - bytes_read))
59 if not chunk:
60 break
61 bytes_read += len(chunk)
62 yield decompressor.decompress(chunk)
64 # The end offset is derived from compressed_size, so the stream must actually
65 # fill it. When the LZMA stream terminates earlier, the surplus bytes inside
66 # the declared region (which may be a following file) would be carved into the
67 # chunk and skipped during the scan instead of being extracted on their own.
68 consumed = bytes_read - len(decompressor.unused_data)
69 if decompressor.eof and consumed < header.compressed_size:
70 raise InvalidInputFormat(
71 "QCA LZMA stream ends before its declared compressed_size"
72 )
75class QcaLzmaHandler(StructHandler):
76 NAME = "qca_lzma"
77 PATTERNS = [HexString("00 00 00 00 ?? ?? ?? ?? 03 00 02 00 00 10 00 00")]
79 DOC = HandlerDoc(
80 name="QCA LZMA",
81 description=(
82 "Compressed LZMA streams with custom header format found in "
83 "powerline firmware images using Qualcomm Atheros (QCA) chips "
84 "of various vendors (TP-Link, Netgear, Trendnet, etc.). The "
85 "images use the NVM format and contain the compressed firmware "
86 "executable."
87 ),
88 handler_type=HandlerType.COMPRESSION,
89 vendor="Qualcomm Atheros",
90 references=[
91 Reference(
92 title="open-plc-utils",
93 url="https://github.com/qca/open-plc-utils",
94 ),
95 ],
96 limitations=[],
97 )
99 C_DEFINITIONS = C_DEFINITIONS
100 HEADER_STRUCT = "qca_lzma_header_t"
101 EXTRACTOR = QcaLzmaExtractor()
102 PATTERN_MATCH_OFFSET = 8
104 def calculate_chunk(self, file: File, start_offset: int) -> ValidChunk | None:
105 header = self.parse_header(file, Endian.LITTLE)
106 end_offset = start_offset + header.size + header.compressed_size
107 self._validate_header(header)
108 try:
109 for _ in _decompress(file, header):
110 pass
111 except lzma.LZMAError as exc:
112 raise InvalidInputFormat("LZMA Decompression failed") from exc
114 return ValidChunk(start_offset=start_offset, end_offset=end_offset)
116 @staticmethod
117 def _validate_header(header) -> None:
118 if header.compressed_size == 0:
119 raise InvalidInputFormat(
120 f"Invalid compressed size {header.compressed_size}"
121 )
122 if (
123 header.decompressed_size == 0
124 or header.decompressed_size < header.compressed_size
125 ):
126 raise InvalidInputFormat(
127 f"Invalid decompressed size {header.decompressed_size}"
128 )