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 bz2
2import gzip
3import io
4import lzma
5from dataclasses import dataclass
6from enum import IntEnum
7from pathlib import Path
8from typing import Any
10import zstandard
12from ...file_utils import FileSystem, convert_int32, convert_int64, round_up
13from ...models import (
14 Endian,
15 Extractor,
16 ExtractResult,
17 File,
18 HandlerDoc,
19 HandlerType,
20 HexString,
21 InvalidInputFormat,
22 Reference,
23 StructHandler,
24 StructParser,
25 ValidChunk,
26)
27from .cpio import CPIOEntry, PortableASCIIParser, StrippedCPIOParser
29RPM_SIGNATURE_ALIGNMENT = 8
30RPMSIGTAG_SIZE = 1000 # INT32
31RPMSIGTAG_LONGSIZE = 270 # INT64
32RPMTAG_PAYLOADCOMPRESSOR = 1125
33RPMTAG_PAYLOADFLAGS = 1126
35RPMTAG_FILEMODES = 1030
36RPMTAG_FILERDEVS = 1033
37RPMTAG_FILELINKTOS = 1036
38RPMTAG_DIRINDEXES = 1116
39RPMTAG_BASENAMES = 1117
40RPMTAG_DIRNAMES = 1118
41RPMTAG_LONGFILESIZES = 5008
43RPM_HEADER_MAGIC = b"\x8e\xad\xe8"
44RPM_HEADER_MAX_TAGS = 0xFFFF
45RPM_HEADER_MAX_DATA_SIZE = 0x0FFF_FFFF
46RPM_HEADER_MAX_ARRAY_SIZE = 0x000F_FFFF
49class RPMType(IntEnum):
50 INT16 = 3
51 INT32 = 4
52 INT64 = 5
53 STRING = 6
54 STRING_ARRAY = 8
55 I18NSTRING_ARRAY = 9
58C_DEFINITIONS = """
59 typedef struct rpm_lead {
60 char magic[4]; // ED AB EE DB
61 uint8 major;
62 uint8 minor;
63 uint16 type;
64 uint16 archnum;
65 char name[66];
66 uint16 osnum;
67 uint16 signature_type;
68 char reserved[16];
69 } rpm_lead_t;
71 // Shared layout for both the Signature header and the Main header.
72 // The variable-length index entries (nindex * 16 bytes) and data
73 // section (hsize bytes) follow immediately after this fixed part.
74 typedef struct rpm_header {
75 char magic[3]; // 8E AD E8
76 uint8 version;
77 char reserved[4];
78 uint32 nindex;
79 uint32 hsize;
80 } rpm_header_t;
82 typedef struct rpm_index_entry {
83 uint32 tag;
84 uint32 type;
85 uint32 offset; // offset into the header's data section
86 uint32 count;
87 } rpm_index_entry_t;
89 typedef struct rpm_cstring {
90 char value[];
91 } rpm_cstring_t;
92 """
95class RPMParser:
96 @dataclass(frozen=True)
97 class HeaderSection:
98 header: Any
99 entries: dict[int, Any]
100 offset: int
101 data_offset: int
102 size: int
104 @property
105 def end_offset(self) -> int:
106 return self.offset + self.size
108 def __init__(self, file: File, start_offset: int):
109 self._file = file
110 self._start_offset = start_offset
111 self._struct_parser = StructParser(C_DEFINITIONS)
113 self._lead: Any | None = None
114 self._signature: RPMParser.HeaderSection | None = None
115 self._main_header: RPMParser.HeaderSection | None = None
116 self._package_size: int | None = None
117 self.compressor: str = "none"
119 def parse(self):
120 self._file.seek(self._start_offset, io.SEEK_SET)
121 self._lead = self._struct_parser.cparser_be.rpm_lead_t(self._file)
123 if self._lead is None:
124 raise InvalidInputFormat("RPM lead is missing or malformed")
126 self._signature = self._read_header_section(
127 offset=self._start_offset + len(self._lead),
128 alignment=RPM_SIGNATURE_ALIGNMENT,
129 )
131 size_entry = self._signature.entries.get(
132 RPMSIGTAG_LONGSIZE
133 ) or self._signature.entries.get(RPMSIGTAG_SIZE)
134 if not size_entry:
135 raise InvalidInputFormat(
136 "RPM signature header must contain a SIZE or LONGSIZE tag"
137 )
139 self._package_size = self._read_entry_integer(self._signature, size_entry)
141 self._main_header = self._read_header_section(offset=self._signature.end_offset)
143 if self.end_offset < self.payload_offset:
144 raise InvalidInputFormat(
145 "RPM package size does not include the main header"
146 )
147 if self.end_offset > len(self._file):
148 raise InvalidInputFormat("RPM package extends beyond the end of the file")
150 compressor_entry = self._main_header.entries.get(RPMTAG_PAYLOADCOMPRESSOR)
151 if compressor_entry:
152 self.compressor = self._read_cstring_entry(
153 self._main_header, compressor_entry
154 )
155 else:
156 # if RPMTAG_PAYLOADCOMPRESSOR is absent on v3/v4 rpm file, the default compression is gzip
157 # while w0.ufdio (none compression) packages omit the tag and set PAYLOADFLAGS = "0" instead.
158 payload_flags_entry = self._main_header.entries.get(RPMTAG_PAYLOADFLAGS)
159 payload_flags = (
160 self._read_cstring_entry(self._main_header, payload_flags_entry)
161 if payload_flags_entry
162 else None
163 )
164 self.compressor = "none" if payload_flags == "0" else "gzip"
166 def _read_header_section(self, offset: int, alignment: int = 1) -> HeaderSection:
167 header_size = self._struct_parser.cparser_be.rpm_header_t.size
168 if offset < self._start_offset or offset + header_size > len(self._file):
169 raise InvalidInputFormat("RPM header extends beyond the end of the file")
171 self._file.seek(offset, io.SEEK_SET)
172 header = self._struct_parser.cparser_be.rpm_header_t(self._file)
173 if (
174 bytes(header.magic) != RPM_HEADER_MAGIC
175 or header.version != 1
176 or bytes(header.reserved) != b"\x00" * 4
177 ):
178 raise InvalidInputFormat("Invalid RPM header record")
179 if header.nindex > RPM_HEADER_MAX_TAGS:
180 raise InvalidInputFormat("RPM header contains too many index entries")
181 if header.hsize > RPM_HEADER_MAX_DATA_SIZE:
182 raise InvalidInputFormat("RPM header data section is too large")
184 entry_size = self._struct_parser.cparser_be.rpm_index_entry_t.size
185 entry_table_size = header.nindex * entry_size
186 data_offset = offset + len(header) + entry_table_size
187 section_size = len(header) + round_up(
188 header.hsize + entry_table_size, alignment
189 )
190 if offset + section_size > len(self._file):
191 raise InvalidInputFormat("RPM header extends beyond the end of the file")
193 entries = {
194 entry.tag: entry
195 for entry in (
196 self._struct_parser.cparser_be.rpm_index_entry_t(self._file)
197 for _ in range(header.nindex)
198 )
199 }
200 if any(entry.offset > header.hsize for entry in entries.values()):
201 raise InvalidInputFormat("RPM index entry points outside its header")
203 return self.HeaderSection(
204 header=header,
205 entries=entries,
206 offset=offset,
207 data_offset=data_offset,
208 size=section_size,
209 )
211 def _read_entry_integer(self, section: HeaderSection, entry: Any) -> int:
212 expected_type = (
213 RPMType.INT64 if entry.tag == RPMSIGTAG_LONGSIZE else RPMType.INT32
214 )
215 size = 8 if expected_type == RPMType.INT64 else 4
216 if entry.type != expected_type or entry.count != 1:
217 raise InvalidInputFormat("RPM package size tag has an invalid type")
218 self._validate_entry_bounds(section, entry, size)
219 self._file.seek(section.data_offset + entry.offset, io.SEEK_SET)
220 if entry.tag == RPMSIGTAG_LONGSIZE:
221 return convert_int64(self._file.read(8), Endian.BIG)
222 return convert_int32(self._file.read(4), Endian.BIG)
224 def _read_cstring_entry(self, section: HeaderSection, entry: Any) -> str:
225 if entry.type != RPMType.STRING or entry.count != 1:
226 raise InvalidInputFormat("RPM string tag has an invalid type")
227 start = section.data_offset + entry.offset
228 end = section.data_offset + section.header.hsize
229 terminator = self._file.find(b"\x00", start, end)
230 if terminator < 0:
231 raise InvalidInputFormat("RPM string is not null-byte terminated")
232 try:
233 return self._file[start:terminator].decode("ascii")
234 except UnicodeDecodeError as e:
235 raise InvalidInputFormat("RPM string is not ASCII") from e
237 def _read_array_entry(self, section: HeaderSection, entry: Any) -> list:
238 """Read a fixed-count array tag value, dispatching on the entry's type."""
239 if entry.count > RPM_HEADER_MAX_ARRAY_SIZE:
240 raise InvalidInputFormat("RPM tag array contains too many values")
241 self._file.seek(section.data_offset + entry.offset, io.SEEK_SET)
242 # this line avoid false pylance errors
243 cparser: Any = self._struct_parser.cparser_be
244 match entry.type:
245 case RPMType.INT16:
246 self._validate_entry_bounds(section, entry, entry.count * 2)
247 return list(cparser.uint16[entry.count](self._file))
248 case RPMType.INT32:
249 self._validate_entry_bounds(section, entry, entry.count * 4)
250 return list(cparser.uint32[entry.count](self._file))
251 case RPMType.INT64:
252 self._validate_entry_bounds(section, entry, entry.count * 8)
253 return list(cparser.uint64[entry.count](self._file))
254 case RPMType.STRING_ARRAY | RPMType.I18NSTRING_ARRAY:
255 values = []
256 end = section.data_offset + section.header.hsize
257 for _ in range(entry.count):
258 start = self._file.tell()
259 terminator = self._file.find(b"\x00", start, end)
260 if terminator < 0:
261 raise InvalidInputFormat(
262 "RPM string array is not null-byte terminated"
263 )
264 values.append(
265 self._file[start:terminator].decode("utf-8", errors="replace")
266 )
267 self._file.seek(terminator + 1, io.SEEK_SET)
268 return values
269 raise InvalidInputFormat(f"Unsupported RPM tag type: {entry.type}")
271 @staticmethod
272 def _validate_entry_bounds(
273 section: HeaderSection, entry: Any, value_size: int
274 ) -> None:
275 if entry.offset + value_size > section.header.hsize:
276 raise InvalidInputFormat("RPM tag value extends outside its header")
278 def build_stripped_entries(self) -> list[CPIOEntry]:
279 """Reconstruct metadata per-file from the main header arrays."""
280 if self._main_header is None:
281 raise InvalidInputFormat("RPM main header has not been parsed")
282 h = self._main_header
283 required_tags = {
284 RPMTAG_BASENAMES,
285 RPMTAG_DIRNAMES,
286 RPMTAG_DIRINDEXES,
287 RPMTAG_LONGFILESIZES,
288 RPMTAG_FILEMODES,
289 RPMTAG_FILELINKTOS,
290 RPMTAG_FILERDEVS,
291 }
292 if missing_tags := required_tags.difference(h.entries):
293 raise InvalidInputFormat(
294 f"RPM stripped payload is missing metadata tags: {sorted(missing_tags)}"
295 )
297 file_names: list[str] = self._read_array_entry(h, h.entries[RPMTAG_BASENAMES])
298 dirnames: list[str] = self._read_array_entry(h, h.entries[RPMTAG_DIRNAMES])
299 dirindexes: list[int] = self._read_array_entry(h, h.entries[RPMTAG_DIRINDEXES])
300 file_sizes: list[int] = self._read_array_entry(
301 h, h.entries[RPMTAG_LONGFILESIZES]
302 )
303 modes: list[int] = self._read_array_entry(h, h.entries[RPMTAG_FILEMODES])
304 links: list[str] = self._read_array_entry(h, h.entries[RPMTAG_FILELINKTOS])
305 rdevs: list[int] = self._read_array_entry(h, h.entries[RPMTAG_FILERDEVS])
306 file_count = len(file_names)
307 if any(
308 len(values) != file_count
309 for values in (dirindexes, file_sizes, modes, links, rdevs)
310 ):
311 raise InvalidInputFormat(
312 "RPM per-file metadata arrays have different sizes"
313 )
314 if any(index >= len(dirnames) for index in dirindexes):
315 raise InvalidInputFormat("RPM directory index is out of range")
317 self._file.seek(self.payload_offset, io.SEEK_SET)
318 return [
319 CPIOEntry(
320 header=None,
321 path=Path(dirnames[dirindexes[i]]) / file_names[i],
322 size=file_sizes[i],
323 mode=modes[i],
324 rdev=rdevs[i],
325 link=links[i],
326 )
327 for i in range(len(file_names))
328 ]
330 @property
331 def payload_offset(self) -> int:
332 if self._main_header is None:
333 raise InvalidInputFormat("RPM main header has not been parsed")
334 return self._main_header.end_offset
336 @property
337 def end_offset(self) -> int:
338 if self._main_header is None or self._package_size is None:
339 raise InvalidInputFormat("RPM package has not been parsed")
340 return self._main_header.offset + self._package_size
342 @property
343 def has_stripped_payload(self) -> bool:
344 """Stripped CPIO is used whenever LONGFILESIZES is needed (any file >4 GiB)."""
345 if self._main_header is None:
346 raise InvalidInputFormat("RPM main header has not been parsed")
347 return RPMTAG_LONGFILESIZES in self._main_header.entries
350class RPMExtractor(Extractor):
351 def extract(self, inpath: Path, outdir: Path) -> ExtractResult:
352 fs = FileSystem(outdir)
353 with File.from_path(inpath) as file:
354 parser = RPMParser(file, 0)
355 parser.parse()
356 file.seek(parser.payload_offset, io.SEEK_SET)
357 with RPMExtractor.open_payload_stream(file, parser.compressor) as decoder:
358 if parser.has_stripped_payload:
359 StrippedCPIOParser(
360 decoder, # pyright: ignore[reportArgumentType]
361 0,
362 parser.build_stripped_entries(),
363 ).parse(fs)
364 else:
365 PortableASCIIParser(decoder, 0).parse(fs) # pyright: ignore[reportArgumentType]
366 return ExtractResult(reports=fs.problems)
368 @staticmethod
369 def open_payload_stream(file: File, compressor: str):
370 """Return a forward-only decompressing reader over the RPM payload.
372 The caller must have already positioned `file` at the payload start.
373 """
374 match compressor:
375 case "gzip":
376 return gzip.GzipFile(fileobj=file, mode="rb")
377 case "bzip2":
378 return bz2.BZ2File(file, mode="rb") # pyright: ignore[reportArgumentType]
379 case "xz" | "lzma":
380 return lzma.LZMAFile(file, mode="rb") # pyright: ignore[reportArgumentType]
381 case "zstd":
382 return zstandard.open(file, mode="rb") # pyright: ignore[reportArgumentType]
383 case "none":
384 return file
385 raise InvalidInputFormat(f"Unsupported RPM payload compressor: {compressor}")
388class RPMHandler(StructHandler):
389 NAME = "rpm"
390 PATTERNS = [
391 HexString(
392 "ED AB EE DB (03 | 04) ?? 00 (00 | 01) ?? ?? [65] 00 ?? ?? 00 05"
393 ) # RPM lead magic + major version (03 or 04) + minor version (?) + type (0x00 or 0x01) + name + signature type (0x5)
394 ]
395 EXTRACTOR = RPMExtractor()
396 C_DEFINITIONS = C_DEFINITIONS
397 HEADER_STRUCT = "rpm_header_t"
399 DOC = HandlerDoc(
400 name="RPM",
401 description="RPM (Red Hat Package Manager) is a package archive format used by Red Hat-based Linux distributions. An RPM file contains metadata (signature, header) and a compressed cpio archive as payload.",
402 handler_type=HandlerType.ARCHIVE,
403 vendor="Red Hat",
404 references=[
405 Reference(
406 title="RPM File Format",
407 url="https://rpm.org/docs/4.20.x/manual/format_v4.html",
408 ),
409 Reference(
410 title="RPM Package Manager",
411 url="https://en.wikipedia.org/wiki/RPM_Package_Manager",
412 ),
413 ],
414 limitations=[],
415 )
417 def calculate_chunk(self, file: File, start_offset: int) -> ValidChunk | None:
418 parser = RPMParser(file, start_offset)
419 parser.parse()
420 return ValidChunk(
421 start_offset=start_offset,
422 end_offset=parser.end_offset,
423 )