1from __future__ import annotations
2
3import io
4from dataclasses import dataclass
5from enum import IntEnum
6from pathlib import Path
7from struct import Struct
8from typing import TYPE_CHECKING
9
10from unblob.file_utils import Endian, FileSystem, InvalidInputFormat, StructParser
11from unblob.models import (
12 Extractor,
13 File,
14 Handler,
15 HandlerDoc,
16 HandlerType,
17 HexString,
18 ValidChunk,
19)
20
21if TYPE_CHECKING:
22 from collections.abc import Iterable
23
24
25C_DEFINITIONS = r"""
26 typedef struct cpkg_header {
27 char magic[4]; // "CPKG"
28 uint32 unknown; // 0x01_00_00_00 in all samples (maybe version)
29 uint32 first_entry_offset;
30 } cpkg_header_t;
31
32 typedef struct fpkg_header {
33 char magic[4]; // "FPKG"
34 uint32 unknown; // 0x01_00_00_00 in all samples
35 uint32 first_entry_offset;
36 uint32 unknown2; // only 0x00_00_00_01 observed
37 uint32 name_len; // length of model name field
38 char name[name_len];
39 } fpkg_header_t;
40
41 typedef struct file_header {
42 uint32 header_len; // only 0x1C (28) observed
43 uint16 type; // probably file type
44 uint16 unknown; // observed values: 0x0 and 0x6874
45 uint32 file_size; // size of file data
46 char filename[];
47 } file_header_t;
48"""
49CPKG_HEADER = "cpkg_header_t"
50FILE_HEADER = "file_header_t"
51CPKG_HEADER_SIZE = 12
52FILE_HEADER_SIZE = 0x1C
53# offset of the filename within a file header (header_len + type + unknown + file_size)
54FILE_HEADER_NAME_OFFSET = 12
55# space left for the filename plus its terminator inside the fixed header
56MAX_FILENAME_SIZE = FILE_HEADER_SIZE - FILE_HEADER_NAME_OFFSET
57VALID_MAGICS = {b"FPKG", b"CPKG"}
58
59
60@dataclass
61class FileHeader(Struct):
62 header_len: int
63 type: int
64 start_offset: int
65 file_size: int
66 filename: str
67 total_size: int
68
69
70class FileType(IntEnum):
71 REGULAR_FILE = 0x100
72 UNKNOWN = 0x101 # NOTE: not observed in the wild, but makes sense in sequence
73 CHECKSUM = 0x102
74 SIGNATURE = 0x103
75
76
77class FPKGParser:
78 def __init__(self, file: File, start_offset: int = 0):
79 self.file = file
80 self.start_offset = start_offset
81 self.struct_parser = StructParser(C_DEFINITIONS)
82 self.header = self.struct_parser.parse(CPKG_HEADER, file, Endian.BIG)
83 self._validate_header(self.header)
84 self.file_data_offset = start_offset + self.header.first_entry_offset
85
86 def iter_entries(self) -> Iterable[FileHeader]:
87 current_offset = self.file_data_offset
88 while current_offset < self.file.size():
89 self.file.seek(current_offset, io.SEEK_SET)
90 entry = self.struct_parser.parse(FILE_HEADER, self.file, Endian.BIG)
91 self._validate_file_header(entry)
92 yield FileHeader(
93 header_len=entry.header_len,
94 file_size=entry.file_size,
95 start_offset=current_offset + entry.header_len,
96 type=entry.type,
97 filename=entry.filename.rstrip().decode("utf-8", errors="replace"),
98 total_size=entry.file_size + entry.header_len,
99 )
100 current_offset += entry.header_len + entry.file_size
101
102 @staticmethod
103 def _validate_file_header(file_header) -> None:
104 if file_header.header_len != FILE_HEADER_SIZE:
105 raise InvalidInputFormat(
106 f"Invalid file header length: {file_header.header_len}"
107 )
108 if len(file_header.filename) >= MAX_FILENAME_SIZE:
109 raise InvalidInputFormat(
110 "Filename not terminated within file header: "
111 f"{file_header.filename[:MAX_FILENAME_SIZE]}..."
112 )
113 if not file_header.filename.isascii():
114 raise InvalidInputFormat(f"Invalid filename: {file_header.filename}")
115 try:
116 FileType(file_header.type)
117 except ValueError as e:
118 raise InvalidInputFormat(f"Invalid file type: {file_header.type}") from e
119
120 @staticmethod
121 def _validate_header(header) -> None:
122 if header.magic not in VALID_MAGICS:
123 raise InvalidInputFormat("Invalid magic")
124 if header.first_entry_offset < CPKG_HEADER_SIZE:
125 raise InvalidInputFormat(
126 f"Invalid first entry offset: {header.first_entry_offset}"
127 )
128
129
130class FPKGExtractor(Extractor):
131 def extract(self, inpath: Path, outdir: Path):
132 fs = FileSystem(outdir)
133 with File.from_path(inpath) as file:
134 parser = FPKGParser(file)
135 for entry in parser.iter_entries():
136 fs.carve(
137 Path(entry.filename), file, entry.start_offset, entry.file_size
138 )
139
140
141class FPKGHandler(Handler):
142 NAME = "fpkg"
143 PATTERNS = [HexString("(43 | 46) 50 4B 47 01 00 00 00")] # (C | F) P K G
144 # FPKG and CPKG headers are compatible if we only care about the "first_entry_offset"
145 EXTRACTOR = FPKGExtractor()
146 DOC = HandlerDoc(
147 name="D-Link FPKG",
148 description="CPKG and FPKG are archive formats used in D-Link DFL firewall firmware",
149 handler_type=HandlerType.ARCHIVE,
150 vendor="D-Link",
151 references=[],
152 limitations=[],
153 )
154
155 def calculate_chunk(self, file: File, start_offset: int) -> ValidChunk | None:
156 parser = FPKGParser(file, start_offset)
157 entries = list(parser.iter_entries())
158 if not entries:
159 raise InvalidInputFormat("No valid entries found")
160
161 end_offset = parser.file_data_offset + sum(e.total_size for e in entries)
162 return ValidChunk(start_offset=start_offset, end_offset=end_offset)