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 io
2import os
3import stat
4from pathlib import Path
6import attrs
7from structlog import get_logger
9from ...file_utils import (
10 Endian,
11 FileSystem,
12 InvalidInputFormat,
13 StructParser,
14 decode_int,
15 iterate_file,
16 round_up,
17 snull,
18)
19from ...models import (
20 Extractor,
21 ExtractResult,
22 File,
23 Handler,
24 HandlerDoc,
25 HandlerType,
26 HexString,
27 Reference,
28 ValidChunk,
29)
30from ...report import ExtractionProblem
32logger = get_logger()
34CPIO_TRAILER_NAME = "TRAILER!!!"
35MAX_LINUX_PATH_LENGTH = 0x1000
37C_TRAILER = 0o00000
38C_ISBLK = 0o60000
39C_ISCHR = 0o20000
40C_ISDIR = 0o40000
41C_ISFIFO = 0o10000
42C_ISSOCK = 0o140000
43C_ISLNK = 0o120000
44C_ISCTG = 0o110000
45C_ISREG = 0o100000
47C_FILE_TYPES = (
48 C_TRAILER,
49 C_ISBLK,
50 C_ISCHR,
51 C_ISDIR,
52 C_ISFIFO,
53 C_ISSOCK,
54 C_ISLNK,
55 C_ISCTG,
56 C_ISREG,
57)
59C_NONE = 0o00000
60C_ISUID = 0o04000
61C_ISGID = 0o02000
62C_ISVTX = 0o01000
63C_ISUID_ISGID = 0o06000
65C_STICKY_BITS = (C_NONE, C_ISUID, C_ISGID, C_ISVTX, C_ISUID_ISGID)
67C_DEFINITIONS = r"""
68 typedef struct old_cpio_header
69 {
70 uint16 c_magic;
71 uint16 c_dev;
72 uint16 c_ino;
73 uint16 c_mode;
74 uint16 c_uid;
75 uint16 c_gid;
76 uint16 c_nlink;
77 uint16 c_rdev;
78 uint16 c_mtimes[2];
79 uint16 c_namesize;
80 uint16 c_filesize[2];
81 } old_cpio_header_t;
83 typedef struct old_ascii_header
84 {
85 char c_magic[6];
86 char c_dev[6];
87 char c_ino[6];
88 char c_mode[6];
89 char c_uid[6];
90 char c_gid[6];
91 char c_nlink[6];
92 char c_rdev[6];
93 char c_mtime[11];
94 char c_namesize[6];
95 char c_filesize[11];
96 } old_ascii_header_t;
98 typedef struct new_ascii_header
99 {
100 char c_magic[6];
101 char c_ino[8];
102 char c_mode[8];
103 char c_uid[8];
104 char c_gid[8];
105 char c_nlink[8];
106 char c_mtime[8];
107 char c_filesize[8];
108 char c_dev_maj[8];
109 char c_dev_min[8];
110 char c_rdev_maj[8];
111 char c_rdev_min[8];
112 char c_namesize[8];
113 char c_chksum[8];
114 } new_ascii_header_t;
115"""
118@attrs.define
119class CPIOEntry:
120 header: object | None
121 size: int
122 mode: int
123 rdev: int
124 path: Path
125 link: str
128class CPIOParserBase:
129 _PAD_ALIGN: int
130 _FILE_PAD_ALIGN: int = 512
131 _STRUCT_PARSER = StructParser(C_DEFINITIONS)
132 HEADER_STRUCT: str
133 entries: list[CPIOEntry]
135 def __init__(
136 self, file: File, start_offset: int, entries: list[CPIOEntry] | None = None
137 ):
138 self.file = file
139 self.start_offset = start_offset
140 self.end_offset = -1
141 self.entries = entries if entries is not None else []
143 def read_entry_header(self) -> CPIOEntry | None:
144 """Parse one entry header + name, return None at EOF."""
145 try:
146 header = self._STRUCT_PARSER.parse(
147 self.HEADER_STRUCT, self.file, Endian.LITTLE
148 )
149 except EOFError:
150 return None
152 c_filesize = self._calculate_file_size(header)
153 c_namesize = self._calculate_name_size(header)
154 self._validate_entry_sizes(c_filesize, c_namesize)
156 padded_header_size = self._pad_header(header, c_namesize)
157 name_padding = padded_header_size - len(header) - c_namesize
158 filename = self._read_entry_filename(c_namesize)
160 if name_padding:
161 self.file.read(name_padding)
163 c_mode = self._calculate_mode(header)
164 file_type = c_mode & 0o770000
165 sticky_bit = c_mode & 0o7000
167 # heuristics 3: check mode field
168 if file_type not in C_FILE_TYPES or sticky_bit not in C_STICKY_BITS:
169 raise InvalidInputFormat("CPIO entry mode is invalid.")
171 return CPIOEntry(
172 header=header,
173 path=Path(filename),
174 size=c_filesize,
175 mode=c_mode,
176 rdev=self._calculate_rdev(header),
177 link="",
178 )
180 @staticmethod
181 def _validate_entry_sizes(c_filesize: int, c_namesize: int) -> None:
182 if c_namesize > MAX_LINUX_PATH_LENGTH:
183 raise InvalidInputFormat("CPIO entry filename is too long.")
184 if c_namesize <= 0:
185 raise InvalidInputFormat("CPIO entry filename size is invalid.")
186 if c_filesize < 0:
187 raise InvalidInputFormat("CPIO entry file size is invalid.")
189 def _read_entry_filename(self, c_namesize: int) -> str:
190 tmp_filename = self.file.read(c_namesize)
191 if len(tmp_filename) != c_namesize or not tmp_filename.endswith(b"\x00"):
192 raise InvalidInputFormat("CPIO entry filename is not null-byte terminated")
193 try:
194 return snull(tmp_filename).decode("utf-8")
195 except UnicodeDecodeError as e:
196 raise InvalidInputFormat from e
198 def parse(self, fs: FileSystem | None = None):
199 while True:
200 entry = self.read_entry_header()
201 if entry is None:
202 break
204 content_padding = self._pad_content(entry.size) - entry.size
206 if entry.path.name == CPIO_TRAILER_NAME:
207 self.file.seek(content_padding, io.SEEK_CUR)
208 break
210 if entry.path.name in ("", ".", ".."):
211 self.file.seek(entry.size + content_padding, io.SEEK_CUR)
212 continue
214 if fs is not None:
215 self.extract_entry(fs, entry)
216 else:
217 self.file.seek(entry.size, io.SEEK_CUR)
218 self.file.seek(content_padding, io.SEEK_CUR)
219 self.end_offset = self._pad_file(self.file.tell())
221 def extract_entry(self, fs: FileSystem, entry: CPIOEntry):
222 # There are cases where CPIO archives have duplicated entries
223 # We then unlink the files to overwrite them and avoid an error.
224 if not stat.S_ISDIR(entry.mode):
225 fs.unlink(entry.path)
227 if stat.S_ISREG(entry.mode):
228 fs.write_chunks(
229 entry.path, iterate_file(self.file, self.file.tell(), entry.size)
230 )
231 elif stat.S_ISLNK(entry.mode):
232 payload_link = snull(self.file.read(entry.size)).decode("utf-8")
233 link_target = entry.link or payload_link
234 fs.create_symlink(src=Path(link_target), dst=entry.path)
235 elif stat.S_ISDIR(entry.mode):
236 fs.mkdir(entry.path, mode=entry.mode & 0o777, parents=True, exist_ok=True)
237 self.file.seek(entry.size, io.SEEK_CUR)
238 elif (
239 stat.S_ISCHR(entry.mode)
240 or stat.S_ISBLK(entry.mode)
241 or stat.S_ISSOCK(entry.mode)
242 ):
243 fs.mknod(entry.path, mode=entry.mode & 0o777, device=entry.rdev)
244 self.file.seek(entry.size, io.SEEK_CUR)
245 else:
246 logger.warning("unknown file type in CPIO archive")
247 self.file.seek(entry.size, io.SEEK_CUR)
249 def verify_checksum_mismatch(
250 self, fs: FileSystem, entry: CPIOEntry, calculated_checksum: int
251 ):
252 if entry.header is None:
253 raise InvalidInputFormat("CPIO checksum metadata is missing")
254 if self.valid_checksum(entry.header, calculated_checksum):
255 return
257 fs.record_problem(
258 ExtractionProblem(
259 problem=(
260 f"CPIO CRC mismatch: expected {decode_int(entry.header.c_chksum, 16):08x}, " # pyright: ignore[reportAttributeAccessIssue]
261 f"got {calculated_checksum:08x}"
262 ),
263 resolution="Extracted anyway.",
264 path=entry.path.as_posix(),
265 )
266 )
268 def _pad_file(self, end_offset: int) -> int:
269 """CPIO archives can have a 512 bytes block padding at the end."""
270 self.file.seek(end_offset, io.SEEK_SET)
271 padded_end_offset = self.start_offset + round_up(
272 size=end_offset - self.start_offset, alignment=self._FILE_PAD_ALIGN
273 )
274 padding_size = padded_end_offset - end_offset
276 if self.file.read(padding_size) == bytes([0]) * padding_size:
277 return padded_end_offset
279 return end_offset
281 @classmethod
282 def _pad_header(cls, header, c_namesize: int) -> int:
283 return round_up(len(header) + c_namesize, cls._PAD_ALIGN)
285 @classmethod
286 def _pad_content(cls, c_filesize: int) -> int:
287 """Pad header and content with _PAD_ALIGN bytes."""
288 return round_up(c_filesize, cls._PAD_ALIGN)
290 @staticmethod
291 def _calculate_file_size(header) -> int:
292 raise NotImplementedError
294 @staticmethod
295 def _calculate_name_size(header) -> int:
296 raise NotImplementedError
298 @staticmethod
299 def _calculate_mode(header) -> int:
300 raise NotImplementedError
302 @staticmethod
303 def _calculate_rdev(header) -> int:
304 raise NotImplementedError
306 def valid_checksum(self, header, calculated_checksum: int) -> bool: # noqa: ARG002
307 return True
310class BinaryCPIOParser(CPIOParserBase):
311 _PAD_ALIGN = 2
313 HEADER_STRUCT = "old_cpio_header_t"
315 @staticmethod
316 def _calculate_file_size(header) -> int:
317 return header.c_filesize[0] << 16 | header.c_filesize[1]
319 @staticmethod
320 def _calculate_name_size(header) -> int:
321 return header.c_namesize + 1 if header.c_namesize % 2 else header.c_namesize
323 @staticmethod
324 def _calculate_mode(header) -> int:
325 return header.c_mode
327 @staticmethod
328 def _calculate_rdev(header) -> int:
329 return header.c_rdev
332class PortableOldASCIIParser(CPIOParserBase):
333 _PAD_ALIGN = 1
335 HEADER_STRUCT = "old_ascii_header_t"
337 @staticmethod
338 def _calculate_file_size(header) -> int:
339 return decode_int(header.c_filesize, 8)
341 @staticmethod
342 def _calculate_name_size(header) -> int:
343 return decode_int(header.c_namesize, 8)
345 @staticmethod
346 def _calculate_mode(header) -> int:
347 return decode_int(header.c_mode, 8)
349 @staticmethod
350 def _calculate_rdev(header) -> int:
351 return decode_int(header.c_rdev, 8)
354class PortableASCIIParser(CPIOParserBase):
355 _PAD_ALIGN = 4
356 HEADER_STRUCT = "new_ascii_header_t"
358 @staticmethod
359 def _calculate_file_size(header) -> int:
360 return decode_int(header.c_filesize, 16)
362 @staticmethod
363 def _calculate_name_size(header) -> int:
364 return decode_int(header.c_namesize, 16)
366 @staticmethod
367 def _calculate_mode(header) -> int:
368 return decode_int(header.c_mode, 16)
370 @staticmethod
371 def _calculate_rdev(header) -> int:
372 return os.makedev(
373 decode_int(header.c_rdev_maj, 16), decode_int(header.c_rdev_min, 16)
374 )
377class PortableASCIIWithCRCParser(PortableASCIIParser):
378 def extract_entry(self, fs: FileSystem, entry: CPIOEntry):
379 if not stat.S_ISDIR(entry.mode):
380 fs.unlink(entry.path)
382 calculated_checksum = 0
384 if stat.S_ISREG(entry.mode):
385 with fs.open(entry.path, "wb+") as output:
386 for chunk in iterate_file(self.file, self.file.tell(), entry.size):
387 calculated_checksum += sum(chunk)
388 output.write(chunk)
389 elif stat.S_ISLNK(entry.mode):
390 content = bytearray()
391 for chunk in iterate_file(self.file, self.file.tell(), entry.size):
392 calculated_checksum += sum(chunk)
393 content.extend(chunk)
394 link_target = snull(bytes(content)).decode("utf-8")
395 fs.create_symlink(src=Path(link_target), dst=entry.path)
396 elif stat.S_ISDIR(entry.mode):
397 fs.mkdir(entry.path, mode=entry.mode & 0o777, parents=True, exist_ok=True)
398 calculated_checksum += self._consume_checksum(entry.size)
399 elif (
400 stat.S_ISCHR(entry.mode)
401 or stat.S_ISBLK(entry.mode)
402 or stat.S_ISSOCK(entry.mode)
403 ):
404 fs.mknod(entry.path, mode=entry.mode & 0o777, device=entry.rdev)
405 calculated_checksum += self._consume_checksum(entry.size)
406 else:
407 logger.warning("unknown file type in CPIO archive")
408 calculated_checksum += self._consume_checksum(entry.size)
410 self.verify_checksum_mismatch(fs, entry, calculated_checksum & 0xFF_FF_FF_FF)
412 def _consume_checksum(self, size: int) -> int:
413 return sum(
414 sum(chunk) for chunk in iterate_file(self.file, self.file.tell(), size)
415 )
417 def valid_checksum(self, header, calculated_checksum: int) -> bool:
418 header_checksum = decode_int(header.c_chksum, 16)
419 return header_checksum == calculated_checksum & 0xFF_FF_FF_FF
422class StrippedCPIOParser(CPIOParserBase):
423 """Stripped CPIO variant (magic 07070X) used in RPM 4.12+.
425 File metadata is supplied at construction from the RPM main header; the parser
426 walks the stream forward to extract each entry.
427 """
429 _PAD_ALIGN = 4
430 _MAGIC = b"07070X"
431 _HEADER_SIZE = 14 # 6 magic + 8 file index
433 def parse(self, fs: FileSystem | None = None):
434 header_padding = self._pad_content(self._HEADER_SIZE) - self._HEADER_SIZE
435 while True:
436 magic = self.file.read(6)
437 # Stripped archives terminate with a standard newc TRAILER entry.
438 if magic in (b"070701", b"070702"):
439 self._read_trailer()
440 break
441 if magic != self._MAGIC:
442 raise InvalidInputFormat(
443 f"Bad stripped CPIO magic: {magic} should be 07070X"
444 )
446 entry = self._read_indexed_entry()
447 self.file.seek(header_padding, io.SEEK_CUR)
448 content_padding = self._pad_content(entry.size) - entry.size
450 if entry.path.name in ("", ".", ".."):
451 self.file.seek(entry.size + content_padding, io.SEEK_CUR)
452 continue
454 if fs is not None:
455 self.extract_entry(fs, entry)
456 else:
457 self.file.seek(entry.size, io.SEEK_CUR)
458 self.file.seek(content_padding, io.SEEK_CUR)
459 self.end_offset = self._pad_file(self.file.tell())
461 def _read_trailer(self) -> None:
462 header_rest = self.file.read(104)
463 if len(header_rest) != 104:
464 raise InvalidInputFormat("Truncated stripped CPIO trailer")
465 name_size = decode_int(header_rest[88:96], 16)
466 self._validate_entry_sizes(0, name_size)
467 trailer_name = self.file.read(name_size)
468 if trailer_name != b"TRAILER!!!\x00":
469 raise InvalidInputFormat("Invalid stripped CPIO trailer")
470 trailer_size = 110 + name_size
471 trailer_padding = round_up(trailer_size, self._PAD_ALIGN) - trailer_size
472 self.file.seek(trailer_padding, io.SEEK_CUR)
474 def _read_indexed_entry(self) -> CPIOEntry:
475 file_index_bytes = self.file.read(8)
476 if len(file_index_bytes) != 8:
477 raise InvalidInputFormat("Truncated stripped CPIO file index")
478 try:
479 return self.entries[int(file_index_bytes, 16)]
480 except (IndexError, ValueError) as e:
481 raise InvalidInputFormat("Invalid stripped CPIO file index") from e
484class _CPIOExtractorBase(Extractor):
485 PARSER: type[CPIOParserBase]
487 def extract(self, inpath: Path, outdir: Path) -> ExtractResult | None:
488 fs = FileSystem(outdir)
490 with File.from_path(inpath) as file:
491 parser = self.PARSER(file, 0)
492 parser.parse(fs)
493 return ExtractResult(reports=fs.problems)
496class BinaryCPIOExtractor(_CPIOExtractorBase):
497 PARSER = BinaryCPIOParser
500class PortableOldASCIIExtractor(_CPIOExtractorBase):
501 PARSER = PortableOldASCIIParser
504class PortableASCIIExtractor(_CPIOExtractorBase):
505 PARSER = PortableASCIIParser
508class PortableASCIIWithCRCExtractor(_CPIOExtractorBase):
509 PARSER = PortableASCIIWithCRCParser
512class _CPIOHandlerBase(Handler):
513 """A common base for all CPIO formats.
515 The format should be parsed the same, there are small differences how to calculate
516 file and filename sizes padding and conversion from octal / hex.
517 """
519 EXTRACTOR: _CPIOExtractorBase
521 def calculate_chunk(self, file: File, start_offset: int) -> ValidChunk | None:
522 parser = self.EXTRACTOR.PARSER(file, start_offset)
523 parser.parse()
524 return ValidChunk(
525 start_offset=start_offset,
526 end_offset=parser.end_offset,
527 )
530class BinaryHandler(_CPIOHandlerBase):
531 NAME = "cpio_binary"
532 PATTERNS = [HexString("c7 71 // (default, bin, hpbin)")]
534 EXTRACTOR = BinaryCPIOExtractor()
536 DOC = HandlerDoc(
537 name="CPIO (binary)",
538 description="CPIO (Copy In, Copy Out) is an archive file format used for bundling files and directories along with their metadata. It is commonly used in Unix-like systems for creating backups or transferring files, and supports various encoding formats including binary and ASCII.",
539 handler_type=HandlerType.ARCHIVE,
540 vendor=None,
541 references=[
542 Reference(
543 title="GNU CPIO Manual",
544 url="https://www.gnu.org/software/cpio/manual/cpio.html",
545 ),
546 ],
547 limitations=[],
548 )
551class PortableOldASCIIHandler(_CPIOHandlerBase):
552 NAME = "cpio_portable_old_ascii"
554 PATTERNS = [HexString("30 37 30 37 30 37 // 07 07 07")]
556 EXTRACTOR = PortableOldASCIIExtractor()
558 DOC = HandlerDoc(
559 name="CPIO (portable old ASCII)",
560 description="CPIO (Copy In, Copy Out) is an archive file format used for bundling files and directories along with their metadata. It is commonly used in Unix-like systems for creating backups or transferring files, and supports various encoding formats including binary and ASCII.",
561 handler_type=HandlerType.ARCHIVE,
562 vendor=None,
563 references=[
564 Reference(
565 title="GNU CPIO Manual",
566 url="https://www.gnu.org/software/cpio/manual/cpio.html",
567 ),
568 ],
569 limitations=[],
570 )
573class PortableASCIIHandler(_CPIOHandlerBase):
574 NAME = "cpio_portable_ascii"
575 PATTERNS = [HexString("30 37 30 37 30 31 // 07 07 01 (newc)")]
577 EXTRACTOR = PortableASCIIExtractor()
579 DOC = HandlerDoc(
580 name="CPIO (portable ASCII)",
581 description="CPIO (Copy In, Copy Out) is an archive file format used for bundling files and directories along with their metadata. It is commonly used in Unix-like systems for creating backups or transferring files, and supports various encoding formats including binary and ASCII.",
582 handler_type=HandlerType.ARCHIVE,
583 vendor=None,
584 references=[
585 Reference(
586 title="GNU CPIO Manual",
587 url="https://www.gnu.org/software/cpio/manual/cpio.html",
588 ),
589 ],
590 limitations=[],
591 )
594class PortableASCIIWithCRCHandler(_CPIOHandlerBase):
595 NAME = "cpio_portable_ascii_crc"
596 PATTERNS = [HexString("30 37 30 37 30 32 // 07 07 02")]
598 EXTRACTOR = PortableASCIIWithCRCExtractor()
600 DOC = HandlerDoc(
601 name="CPIO (portable ASCII CRC)",
602 description="CPIO (Copy In, Copy Out) is an archive file format used for bundling files and directories along with their metadata. It is commonly used in Unix-like systems for creating backups or transferring files, and supports various encoding formats including binary and ASCII.",
603 handler_type=HandlerType.ARCHIVE,
604 vendor=None,
605 references=[
606 Reference(
607 title="GNU CPIO Manual",
608 url="https://www.gnu.org/software/cpio/manual/cpio.html",
609 ),
610 ],
611 limitations=[],
612 )