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
1from __future__ import annotations
3import io
4import os
5import stat
6import struct
7from enum import IntEnum, unique
8from pathlib import Path
10from structlog import get_logger
12from ...file_utils import (
13 Endian,
14 FileSystem,
15 InvalidInputFormat,
16 iterate_file,
17 read_until_past,
18 round_up,
19)
20from ...models import (
21 Extractor,
22 ExtractResult,
23 File,
24 HandlerDoc,
25 HandlerType,
26 HexString,
27 Reference,
28 StructHandler,
29 ValidChunk,
30)
32logger = get_logger()
35STRING_ALIGNMENT = 16
36MAX_LINUX_PATH_LENGTH = 0xFF
37MAX_UINT32 = 0x100000000
40WORLD_RW = 0o666
41WORLD_RWX = 0o777
42ROMFS_HEADER_SIZE = 512
43ROMFS_SIGNATURE = b"-rom1fs-"
46@unique
47class FSType(IntEnum):
48 HARD_LINK = 0
49 DIRECTORY = 1
50 FILE = 2
51 SYMLINK = 3
52 BLOCK_DEV = 4
53 CHAR_DEV = 5
54 SOCKET = 6
55 FIFO = 7
58def valid_checksum(content: bytes) -> bool:
59 """Apply a RomFS checksum and returns whether it's valid or not."""
60 total = 0
62 # unalign content will lead to unpacking errors down the line
63 if len(content) % 4 != 0:
64 return False
66 for i in range(0, len(content), 4):
67 total = (total + struct.unpack(">L", content[i : i + 4])[0]) % MAX_UINT32
68 return total == 0
71def get_string(file: File) -> bytes:
72 """Read a 16 bytes aligned, null terminated string."""
73 filename = b""
74 counter = 0
75 while b"\x00" not in filename and counter < MAX_LINUX_PATH_LENGTH:
76 filename += file.read(STRING_ALIGNMENT)
77 counter += STRING_ALIGNMENT
78 return filename.rstrip(b"\x00")
81class FileHeader:
82 addr: int
83 next_filehdr: int
84 spec_info: int
85 fs_type: FSType
86 executable: bool
87 size: int
88 checksum: int
89 filename: bytes
90 depth: int = -1
91 parent: FileHeader | None = None
92 start_offset: int
93 end_offset: int
94 file: File
96 def __init__(self, addr: int, file: File):
97 self.addr = addr
98 fs_typeexec_next = struct.unpack(">L", file.read(4))[0]
99 self.next_filehdr = fs_typeexec_next & ~0b1111
100 self.fs_type = FSType(fs_typeexec_next & 0b0111)
101 self.executable = fs_typeexec_next & 0b1000
102 self.spec_info = struct.unpack(">I", file.read(4))[0]
103 self.size = struct.unpack(">I", file.read(4))[0]
104 self.checksum = struct.unpack(">I", file.read(4))[0]
105 self.filename = get_string(file)
106 self.start_offset = file.tell()
107 self.file = file
109 def valid_checksum(self) -> bool:
110 current_position = self.file.tell()
111 try:
112 self.file.seek(self.addr, io.SEEK_SET)
113 filename_len = len(self.filename)
114 header_size = 16 + round_up(filename_len, 16)
115 return valid_checksum(self.file.read(header_size))
116 finally:
117 self.file.seek(current_position, io.SEEK_SET)
119 def content(self) -> bytes:
120 """Return the file content. Applicable to files and symlinks."""
121 self.validate_content_bounds()
122 try:
123 self.file.seek(self.start_offset, io.SEEK_SET)
124 return self.file.read(self.size)
125 finally:
126 self.file.seek(self.start_offset, io.SEEK_SET)
128 def validate_content_bounds(self) -> None:
129 """Raise when the inode content extends past the ROMFS image."""
130 if self.start_offset + self.size > self.file.size():
131 raise RomFSError("Inode size extends past the end of the file")
133 @property
134 def mode(self) -> int:
135 """Permission mode.
137 It is assumed to be world readable if executable bit is set,
138 and world executable otherwise. Handle mode for both block
139 device and character devices too.
140 """
141 mode = WORLD_RWX if self.executable else WORLD_RW
142 mode |= stat.S_IFBLK if self.fs_type == FSType.BLOCK_DEV else 0x0
143 mode |= stat.S_IFCHR if self.fs_type == FSType.CHAR_DEV else 0x0
144 return mode
146 @property
147 def dev(self) -> int:
148 """Raw device number if block device or character device, zero otherwise."""
149 if self.fs_type in [FSType.BLOCK_DEV, FSType.CHAR_DEV]:
150 major = self.spec_info >> 16
151 minor = self.spec_info & 0xFFFF
152 return os.makedev(major, minor)
153 return 0
155 @property
156 def path(self) -> Path:
157 """Returns the full path of this file, up to the RomFS root."""
158 current_node = self
159 current_path = Path()
160 while current_node is not None:
161 current_path = Path(current_node.filename.decode("utf-8")).joinpath(
162 current_path
163 )
164 current_node = current_node.parent
165 return current_path
167 def __repr__(self):
168 return (
169 f"FileHeader<next_filehdr:{self.next_filehdr}, type:{self.fs_type},"
170 f" executable:{self.executable}, spec_info:{self.spec_info},"
171 f" size:{self.size}, checksum:{self.checksum}, filename:{self.filename}>"
172 )
175class RomFSError(Exception):
176 pass
179class RomFSHeader:
180 signature: bytes
181 full_size: int
182 checksum: int
183 volume_name: bytes
184 eof: int
185 file: File
186 end_offset: int
187 inodes: dict[int, FileHeader]
188 fs: FileSystem
190 def __init__(
191 self,
192 file: File,
193 fs: FileSystem,
194 ):
195 self.file = file
196 self.file.seek(0, io.SEEK_END)
197 self.eof = self.file.tell()
198 self.file.seek(0, io.SEEK_SET)
200 if self.eof < ROMFS_HEADER_SIZE:
201 raise RomFSError("File too small to hold ROMFS")
203 self.signature = self.file.read(8)
204 self.full_size = struct.unpack(">I", self.file.read(4))[0]
205 self.checksum = struct.unpack(">I", self.file.read(4))[0]
206 self.volume_name = get_string(self.file)
207 self.header_end_offset = self.file.tell()
208 self.inodes = {}
210 self.fs = fs
212 def valid_checksum(self) -> bool:
213 current_position = self.file.tell()
214 try:
215 self.file.seek(0, io.SEEK_SET)
216 return valid_checksum(self.file.read(ROMFS_HEADER_SIZE))
217 finally:
218 self.file.seek(current_position, io.SEEK_SET)
220 def validate(self):
221 if self.signature != ROMFS_SIGNATURE:
222 raise RomFSError("Invalid RomFS signature")
223 if self.full_size > self.eof:
224 raise RomFSError("ROMFS size is greater than file size")
225 if not self.valid_checksum():
226 raise RomFSError("Invalid checksum")
228 def is_valid_addr(self, addr):
229 """Validate that an inode address is valid.
231 Inodes addresses must be 16 bytes aligned and placed within
232 the RomFS on file.
233 """
234 return (self.header_end_offset <= addr <= self.eof) and (addr % 16 == 0)
236 def is_recursive(self, addr) -> bool:
237 return addr in self.inodes
239 def recursive_walk(self, addr: int, parent: FileHeader | None = None):
240 while self.is_valid_addr(addr) is True:
241 addr = self.walk_dir(addr, parent)
243 def walk_dir(self, addr: int, parent: FileHeader | None = None):
244 self.file.seek(addr, io.SEEK_SET)
245 file_header = FileHeader(addr, self.file)
246 file_header.parent = parent
248 if not file_header.valid_checksum():
249 raise RomFSError(f"Invalid file CRC at addr {addr:0x}.")
251 logger.debug("walking dir", addr=addr, file=file_header)
253 if file_header.filename not in [b".", b".."]:
254 if (
255 file_header.fs_type == FSType.DIRECTORY
256 and file_header.spec_info != 0x0
257 and not self.is_recursive(addr)
258 ):
259 self.inodes[addr] = file_header
260 self.recursive_walk(file_header.spec_info, file_header)
261 self.inodes[addr] = file_header
262 return file_header.next_filehdr
264 def create_symlink(self, output_path: Path, inode: FileHeader):
265 target_path = Path(inode.content().decode("utf-8"))
266 self.fs.create_symlink(src=target_path, dst=output_path)
268 def create_hardlink(self, output_path: Path, inode: FileHeader):
269 if inode.spec_info in self.inodes:
270 target_path = self.inodes[inode.spec_info].path
271 self.fs.create_hardlink(dst=output_path, src=target_path)
272 else:
273 logger.warning("Invalid hard link target", inode_key=inode.spec_info)
275 def create_inode(self, inode: FileHeader):
276 output_path = inode.path
277 logger.info("dumping inode", inode=inode, output_path=str(output_path))
279 if inode.fs_type == FSType.HARD_LINK:
280 self.create_hardlink(output_path, inode)
281 elif inode.fs_type == FSType.SYMLINK:
282 self.create_symlink(output_path, inode)
283 elif inode.fs_type == FSType.DIRECTORY:
284 self.fs.mkdir(output_path, mode=inode.mode, exist_ok=True)
285 elif inode.fs_type == FSType.FILE:
286 inode.validate_content_bounds()
287 self.fs.write_chunks(
288 output_path,
289 iterate_file(inode.file, inode.start_offset, inode.size),
290 )
291 elif inode.fs_type in [FSType.BLOCK_DEV, FSType.CHAR_DEV]:
292 self.fs.mknod(output_path, mode=inode.mode, device=inode.dev)
293 elif inode.fs_type == FSType.FIFO:
294 self.fs.mkfifo(output_path, mode=inode.mode)
296 def dump_fs(self):
297 def inodes(*inode_types):
298 return sorted(
299 (v for v in self.inodes.values() if v.fs_type in inode_types),
300 key=lambda inode: inode.path,
301 )
303 # order of file object creation is important
304 sorted_inodes = (
305 inodes(FSType.FILE, FSType.DIRECTORY, FSType.FIFO, FSType.SOCKET)
306 + inodes(FSType.BLOCK_DEV, FSType.CHAR_DEV)
307 + inodes(FSType.SYMLINK, FSType.HARD_LINK)
308 )
310 for inode in sorted_inodes:
311 self.create_inode(inode)
313 def __str__(self):
314 return f"signature: {self.signature}\nfull_size: {self.full_size}\nchecksum: {self.checksum}\nvolume_name: {self.volume_name}"
317class RomfsExtractor(Extractor):
318 def extract(self, inpath: Path, outdir: Path):
319 fs = FileSystem(outdir)
320 with File.from_path(inpath) as f:
321 header = RomFSHeader(f, fs)
322 header.validate()
323 header.recursive_walk(header.header_end_offset, None)
324 header.dump_fs()
325 return ExtractResult(reports=fs.problems)
328class RomFSFSHandler(StructHandler):
329 NAME = "romfs"
331 PATTERNS = [
332 # '-rom1fs-'
333 HexString("2D 72 6F 6D 31 66 73 2d")
334 ]
336 C_DEFINITIONS = r"""
337 struct romfs_header {
338 char magic[8];
339 uint32 full_size;
340 uint32 checksum;
341 }
342 """
343 HEADER_STRUCT = "romfs_header"
344 EXTRACTOR = RomfsExtractor()
346 DOC = HandlerDoc(
347 name="RomFS",
348 description="RomFS is a simple, space-efficient, read-only file system format designed for embedded systems. It features 16-byte alignment, minimal metadata overhead, and supports basic file types like directories, files, symlinks, and devices.",
349 handler_type=HandlerType.FILESYSTEM,
350 vendor=None,
351 references=[
352 Reference(
353 title="RomFS Documentation",
354 url="https://www.kernel.org/doc/html/latest/filesystems/romfs.html",
355 ),
356 Reference(
357 title="RomFS Wikipedia",
358 url="https://en.wikipedia.org/wiki/Romfs",
359 ),
360 ],
361 limitations=[],
362 )
364 def calculate_chunk(self, file: File, start_offset: int) -> ValidChunk | None:
365 if not valid_checksum(file.read(512)):
366 raise InvalidInputFormat("Invalid RomFS checksum.")
368 file.seek(-512, io.SEEK_CUR)
370 # Every multi byte value must be in big endian order.
371 header = self.parse_header(file, Endian.BIG)
373 # The zero terminated name of the volume, padded to 16 byte boundary.
374 get_string(file)
376 # seek filesystem size (number of accessible bytes in this fs)
377 # from the actual end of the header
378 file.seek(header.full_size, io.SEEK_CUR)
380 # Another thing to note is that romfs works on file headers and data
381 # aligned to 16 byte boundaries, but most hardware devices and the block
382 # device drivers are unable to cope with smaller than block-sized data.
383 # To overcome this limitation, the whole size of the file system must be
384 # padded to an 1024 byte boundary.
385 read_until_past(file, b"\x00")
387 return ValidChunk(
388 start_offset=start_offset,
389 end_offset=file.tell(),
390 )