This commit is contained in:
2026-08-18 09:20:44 +02:00
commit b826cd721a
522 changed files with 93730 additions and 0 deletions
+764
View File
@@ -0,0 +1,764 @@
#!/usr/bin/env python3
"""Inspect and make assertion-driven EROFS manual-test fixtures."""
from __future__ import annotations
import argparse
import hashlib
import shutil
import struct
from dataclasses import dataclass
from pathlib import Path
SUPER = 1024
MAGIC = 0xE0F5E1E2
FEATURE_COMPAT_SB_CHKSUM = 0x00000001
FEATURE_INCOMPAT_COMPR_CFGS = 0x00000002
FEATURE_INCOMPAT_48BIT = 0x00000080
CRC32C_POLYNOMIAL = 0x82F63B78
CRC32C_INITIAL = 0xFFFFFFFF
KERNEL_CRC32C_SEED = 0x5045B54A
def crc32c(data: bytes | bytearray, initial: int = CRC32C_INITIAL) -> int:
checksum = initial
for byte in data:
checksum ^= byte
for _ in range(8):
checksum = (checksum >> 1) ^ (
CRC32C_POLYNOMIAL if checksum & 1 else 0
)
return checksum & 0xFFFFFFFF
@dataclass(frozen=True)
class Inode:
nid: int
offset: int
inode_format: int
inode_size: int
xattr_size: int
layout: int
mode: int
size: int
start_block_low: int
start_block_high: int
start_block: int
@dataclass(frozen=True)
class DirectoryEntry:
nid: int
offset: int
name_offset: int
end_offset: int
name: bytes
class ErofsImage:
def __init__(self, data: bytearray, source: Path):
self.data = data
self.source = source
if len(data) < SUPER + 128:
raise ValueError(f"{source}: shorter than the EROFS superblock")
@classmethod
def load(cls, path: Path) -> "ErofsImage":
return cls(bytearray(path.read_bytes()), path)
def clone(self) -> "ErofsImage":
return ErofsImage(bytearray(self.data), self.source)
def u16(self, offset: int) -> int:
return struct.unpack_from("<H", self.data, offset)[0]
def u32(self, offset: int) -> int:
return struct.unpack_from("<I", self.data, offset)[0]
def u64(self, offset: int) -> int:
return struct.unpack_from("<Q", self.data, offset)[0]
def put_u16(self, offset: int, value: int) -> None:
struct.pack_into("<H", self.data, offset, value)
def put_u32(self, offset: int, value: int) -> None:
struct.pack_into("<I", self.data, offset, value)
def put_u64(self, offset: int, value: int) -> None:
struct.pack_into("<Q", self.data, offset, value)
@property
def block_bits(self) -> int:
return self.data[SUPER + 12]
@property
def block_size(self) -> int:
if not 9 <= self.block_bits <= 16:
raise ValueError(f"invalid block bits {self.block_bits}")
return 1 << self.block_bits
@property
def checksum_end(self) -> int:
span = self.block_size
if span > SUPER:
span -= SUPER
end = SUPER + span
if end > len(self.data):
raise ValueError(
f"checksum window ends at {end}, image size is {len(self.data)}"
)
return end
@property
def feature_compat(self) -> int:
return self.u32(SUPER + 8)
@property
def feature_incompat(self) -> int:
return self.u32(SUPER + 80)
@property
def blocks(self) -> int:
blocks = self.u32(SUPER + 36)
root8 = self.u64(SUPER + 112)
if self.feature_incompat & FEATURE_INCOMPAT_48BIT and root8 != 0:
blocks |= self.u16(SUPER + 14) << 32
return blocks
@property
def root_nid(self) -> int:
root8 = self.u64(SUPER + 112)
if self.feature_incompat & FEATURE_INCOMPAT_48BIT and root8 != 0:
return root8
return self.u16(SUPER + 14)
def calculated_checksum(self) -> int:
window = bytearray(self.data[SUPER : self.checksum_end])
struct.pack_into("<I", window, 4, 0)
return crc32c(window)
def kernel_calculated_checksum(self) -> int:
return crc32c(
self.data[SUPER + 8 : self.checksum_end], KERNEL_CRC32C_SEED
)
def checksum_valid(self) -> bool:
if not self.feature_compat & FEATURE_COMPAT_SB_CHKSUM:
return True
return self.u32(SUPER + 4) == self.calculated_checksum()
def kernel_checksum_valid(self) -> bool:
if not self.feature_compat & FEATURE_COMPAT_SB_CHKSUM:
return True
return self.u32(SUPER + 4) == self.kernel_calculated_checksum()
def update_checksum(self) -> None:
if not self.feature_compat & FEATURE_COMPAT_SB_CHKSUM:
raise ValueError("image does not advertise superblock checksum")
self.put_u32(SUPER + 4, 0)
self.put_u32(SUPER + 4, self.calculated_checksum())
if not self.checksum_valid():
raise AssertionError("updated checksum does not verify")
if (
self.u32(SUPER) == MAGIC
and self.calculated_checksum() != self.kernel_calculated_checksum()
):
raise AssertionError("canonical and kernel checksum forms differ")
def validate_superblock(self) -> None:
if self.u32(SUPER) != MAGIC:
raise ValueError(f"unexpected magic {self.u32(SUPER):#010x}")
if self.blocks == 0:
raise ValueError("declared block count is zero")
if not self.checksum_valid():
raise ValueError("superblock checksum is invalid")
def inode(self, nid: int) -> Inode:
metadata = self.u32(SUPER + 40) << self.block_bits
offset = metadata + (nid << 5)
if offset > len(self.data) - 32:
raise ValueError(f"nid {nid} maps outside the image at {offset}")
inode_format = self.u16(offset)
inode_size = 64 if inode_format & 1 else 32
if offset > len(self.data) - inode_size:
raise ValueError(f"nid {nid} extended inode is truncated")
xattr_count = self.u16(offset + 2)
xattr_size = 0 if xattr_count == 0 else 12 + 4 * (xattr_count - 1)
size = self.u64(offset + 8) if inode_size == 64 else self.u32(offset + 8)
start_block_low = self.u32(offset + 16)
start_block_high = (
self.u16(offset + 6)
if inode_size == 64
and self.feature_incompat & FEATURE_INCOMPAT_48BIT
else 0
)
return Inode(
nid=nid,
offset=offset,
inode_format=inode_format,
inode_size=inode_size,
xattr_size=xattr_size,
layout=(inode_format >> 1) & 7,
mode=self.u16(offset + 4),
size=size,
start_block_low=start_block_low,
start_block_high=start_block_high,
start_block=start_block_low | (start_block_high << 32),
)
def directory_entries(self, inode: Inode) -> list[DirectoryEntry]:
if inode.size == 0 or inode.size > self.block_size:
raise ValueError("helper resolves only one-block directories")
if inode.layout == 2:
data_offset = inode.offset + inode.inode_size + inode.xattr_size
elif inode.layout == 0:
data_offset = inode.start_block << self.block_bits
else:
raise ValueError(f"unsupported directory layout {inode.layout}")
if data_offset > len(self.data) - inode.size:
raise ValueError("directory data lies outside the image")
first_name_offset = self.u16(data_offset + 8)
if (
first_name_offset < 12
or first_name_offset % 12 != 0
or first_name_offset >= inode.size
):
raise ValueError("invalid first directory name offset")
count = first_name_offset // 12
entries = []
previous = 0
for index in range(count):
entry_offset = data_offset + index * 12
name_offset = self.u16(entry_offset + 8)
end_offset = (
self.u16(entry_offset + 20)
if index + 1 < count
else inode.size
)
if (
name_offset < first_name_offset
or name_offset <= previous
or end_offset <= name_offset
or end_offset > inode.size
):
raise ValueError("invalid directory name offsets")
slot = bytes(
self.data[
data_offset + name_offset : data_offset + end_offset
]
)
name = slot.split(b"\0", 1)[0]
if not name:
raise ValueError("empty directory name")
entries.append(
DirectoryEntry(
nid=self.u64(entry_offset),
offset=entry_offset,
name_offset=name_offset,
end_offset=end_offset,
name=name,
)
)
previous = name_offset
return entries
def resolve_root_entry(self, path: str) -> tuple[Inode, DirectoryEntry]:
name = path.removeprefix("/").encode("ascii")
if not name or b"/" in name:
raise ValueError("helper accepts one root-level path component")
root = self.inode(self.root_nid)
for entry in self.directory_entries(root):
if entry.name == name:
return root, entry
raise ValueError(f"path not found in root directory: {path}")
def save(self, path: Path) -> None:
path.write_bytes(self.data)
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def write_checksums(output: Path) -> None:
paths = sorted(output.glob("*.erofs"))
with (output / "SHA256SUMS").open("w", encoding="ascii") as sums:
for path in paths:
sums.write(f"{sha256(path)} {path.name}\n")
def save_checked(image: ErofsImage, path: Path) -> None:
image.update_checksum()
image.validate_superblock()
image.save(path)
def assert_same_length(original: ErofsImage, changed: ErofsImage) -> None:
if len(original.data) != len(changed.data):
raise AssertionError("fixture mutation changed provider length")
def make_error_fixtures(args: argparse.Namespace) -> None:
output = args.output
if output.exists():
raise ValueError(f"output already exists: {output}")
output.mkdir(parents=True)
plain = ErofsImage.load(args.plain)
compressed = ErofsImage.load(args.compressed)
deflate = ErofsImage.load(args.deflate)
plain.validate_superblock()
compressed.validate_superblock()
deflate.validate_superblock()
shutil.copy2(args.plain, output / "valid-plain.erofs")
shutil.copy2(args.compressed, output / "valid-lz4.erofs")
shutil.copy2(args.deflate, output / "valid-deflate-level1.erofs")
evidence = []
bad_crc = plain.clone()
patch_offset = SUPER + 64
old_byte = bad_crc.data[patch_offset]
bad_crc.data[patch_offset] ^= 0x01
assert_same_length(plain, bad_crc)
if bad_crc.checksum_valid():
raise AssertionError("bad checksum fixture still verifies")
bad_crc.save(output / "bad-super-crc.erofs")
evidence.append(
f"bad-super-crc patch_offset={patch_offset} checksum_offset={SUPER + 4} "
f"old={old_byte:#04x} new={bad_crc.data[patch_offset]:#04x} "
f"coverage={SUPER}:{plain.checksum_end} recomputed=no "
f"provider_length={len(plain.data)}"
)
bad_checksum_field = plain.clone()
old_checksum = bad_checksum_field.u32(SUPER + 4)
bad_checksum_field.put_u32(SUPER + 4, old_checksum ^ 0x00000001)
assert_same_length(plain, bad_checksum_field)
if bad_checksum_field.checksum_valid():
raise AssertionError("altered checksum field still verifies")
bad_checksum_field.save(output / "bad-checksum-field.erofs")
evidence.append(
f"bad-checksum-field field_offset={SUPER + 4} "
f"old={old_checksum:#010x} new={old_checksum ^ 1:#010x} "
f"coverage={SUPER}:{plain.checksum_end} recomputed=no "
f"provider_length={len(plain.data)}"
)
bad_magic = plain.clone()
if bad_magic.u32(SUPER) != MAGIC:
raise AssertionError("unexpected source magic")
bad_magic.put_u32(SUPER, 0x21444142)
assert_same_length(plain, bad_magic)
if bad_magic.checksum_valid() or not bad_magic.kernel_checksum_valid():
raise AssertionError("bad magic checksum qualification failed")
bad_magic.save(output / "bad-magic.erofs")
evidence.append(
f"bad-magic patch_offset={SUPER} old={MAGIC:#010x} new=0x21444142 "
f"crc=unchanged canonical_valid=no kernel_suffix_valid=yes "
f"provider_length={len(plain.data)}"
)
bad_block_size = plain.clone()
old_block_bits = bad_block_size.block_bits
bad_block_size.data[SUPER + 12] = 13
bad_block_size.update_checksum()
assert_same_length(plain, bad_block_size)
bad_block_size.save(output / "bad-block-size.erofs")
evidence.append(
f"bad-block-size field_offset={SUPER + 12} old={old_block_bits} new=13 "
f"coverage={SUPER}:{bad_block_size.checksum_end} crc=recomputed "
f"provider_length={len(plain.data)}"
)
bad_root = plain.clone()
if bad_root.feature_incompat & FEATURE_INCOMPAT_48BIT:
raise AssertionError("plain source unexpectedly has 48-bit feature")
metadata_offset = bad_root.u32(SUPER + 40) << bad_root.block_bits
invalid_root = (
((bad_root.blocks << bad_root.block_bits) - metadata_offset) // 32
+ 1024
)
bad_root.put_u32(
SUPER + 80, bad_root.feature_incompat | FEATURE_INCOMPAT_48BIT
)
bad_root.put_u16(SUPER + 14, 0)
bad_root.put_u64(SUPER + 112, invalid_root)
save_checked(bad_root, output / "bad-root-nid.erofs")
assert_same_length(plain, bad_root)
evidence.append(
f"bad-root-nid feature_offset={SUPER + 80} "
f"blocks_hi_offset={SUPER + 14} rootnid_8b_offset={SUPER + 112} "
f"old_root={plain.root_nid} new_root={invalid_root} "
f"crc=recomputed provider_length={len(plain.data)}"
)
unsupported_feature = plain.clone()
old_incompat = unsupported_feature.feature_incompat
unsupported_feature.put_u32(SUPER + 80, old_incompat | 0x80000000)
save_checked(unsupported_feature, output / "unsupported-feature.erofs")
assert_same_length(plain, unsupported_feature)
evidence.append(
f"unsupported-feature field_offset={SUPER + 80} "
f"old={old_incompat:#010x} new={old_incompat | 0x80000000:#010x} "
f"crc=recomputed provider_length={len(plain.data)}"
)
short_path = output / "truncated-before-super.erofs"
short_path.write_bytes(plain.data[: SUPER - 1])
evidence.append(
f"truncated-before-super source_length={len(plain.data)} "
f"provider_length={SUPER - 1} required_super_offset={SUPER}"
)
empty_path = output / "empty-provider.erofs"
empty_path.write_bytes(b"")
evidence.append(
f"empty-provider source_length={len(plain.data)} provider_length=0"
)
bad_dirent = plain.clone()
_, entry = bad_dirent.resolve_root_entry("/bad-entry.txt")
invalid_nid = (bad_dirent.blocks << bad_dirent.block_bits) // 32 + 1024
bad_dirent.put_u64(entry.offset, invalid_nid)
save_checked(bad_dirent, output / "bad-dirent-nid.erofs")
assert_same_length(plain, bad_dirent)
evidence.append(
f"bad-dirent-nid dirent_offset={entry.offset} old_nid={entry.nid} "
f"new_nid={invalid_nid} crc=recomputed "
f"provider_length={len(plain.data)}"
)
bad_inode = plain.clone()
_, entry = bad_inode.resolve_root_entry("/inode-target.txt")
inode = bad_inode.inode(entry.nid)
if inode.inode_format & 0x8000:
raise AssertionError("reserved i_format bit already set")
bad_inode.put_u16(inode.offset, inode.inode_format | 0x8000)
save_checked(bad_inode, output / "bad-inode-format.erofs")
assert_same_length(plain, bad_inode)
evidence.append(
f"bad-inode-format nid={inode.nid} inode_offset={inode.offset} "
f"i_format={inode.inode_format:#06x}->{inode.inode_format | 0x8000:#06x} "
f"crc=recomputed provider_length={len(plain.data)}"
)
out_of_bounds = plain.clone()
_, entry = out_of_bounds.resolve_root_entry("/plain.bin")
inode = out_of_bounds.inode(entry.nid)
if inode.layout != 0 or inode.size == 0:
raise AssertionError("plain.bin is not nonempty FLAT_PLAIN")
out_of_bounds.put_u32(inode.offset + 16, out_of_bounds.blocks)
save_checked(out_of_bounds, output / "oob-start-block.erofs")
assert_same_length(plain, out_of_bounds)
evidence.append(
f"oob-start-block nid={inode.nid} field_offset={inode.offset + 16} "
f"old={inode.start_block} new={out_of_bounds.blocks} crc=recomputed "
f"provider_length={len(plain.data)}"
)
future = compressed.clone()
if not future.feature_incompat & FEATURE_INCOMPAT_COMPR_CFGS:
raise AssertionError("compressed image has no compression config feature")
available_offset = SUPER + 82
old_algorithms = future.u16(available_offset)
if old_algorithms & 0x8000:
raise AssertionError("future algorithm bit already set")
future.put_u16(available_offset, old_algorithms | 0x8000)
save_checked(future, output / "future-algorithm.erofs")
assert_same_length(compressed, future)
evidence.append(
f"future-algorithm field_offset={available_offset} "
f"old={old_algorithms:#06x} new={old_algorithms | 0x8000:#06x} "
f"crc=recomputed provider_length={len(compressed.data)}"
)
compressed_corrupt = compressed.clone()
corrupt_offset = args.compressed_offset + 32
corrupt_length = 64
if args.compressed_length < 32 + corrupt_length:
raise ValueError("compressed extent is too short for targeted mutation")
if corrupt_offset + corrupt_length > (
args.compressed_offset + args.compressed_length
):
raise ValueError("compressed mutation exceeds selected extent")
if corrupt_offset < compressed_corrupt.checksum_end:
raise ValueError("compressed corruption overlaps checksum window")
if corrupt_offset + corrupt_length > len(compressed_corrupt.data):
raise ValueError("compressed corruption exceeds provider")
old_bytes = bytes(
compressed_corrupt.data[corrupt_offset : corrupt_offset + corrupt_length]
)
if old_bytes == bytes(corrupt_length):
raise ValueError("compressed mutation would not change bytes")
compressed_corrupt.data[
corrupt_offset : corrupt_offset + corrupt_length
] = bytes(corrupt_length)
if not compressed_corrupt.checksum_valid():
raise AssertionError("payload mutation changed superblock checksum")
assert_same_length(compressed, compressed_corrupt)
compressed_corrupt.save(output / "compressed-stream-corrupt.erofs")
evidence.append(
f"compressed-stream-corrupt extent_offset={args.compressed_offset} "
f"extent_length={args.compressed_length} patch_offset={corrupt_offset} "
f"patch_length={corrupt_length} provider_length={len(compressed.data)} "
"crc=unchanged-valid"
)
raw_corrupt = plain.clone()
_, entry = raw_corrupt.resolve_root_entry("/plain.bin")
inode = raw_corrupt.inode(entry.nid)
if inode.layout != 0 or inode.size <= 257:
raise ValueError("plain.bin is not a qualifying FLAT_PLAIN target")
raw_offset = (inode.start_block << raw_corrupt.block_bits) + 257
if raw_offset < raw_corrupt.checksum_end or raw_offset >= len(raw_corrupt.data):
raise ValueError("raw payload patch offset is invalid")
old_byte = raw_corrupt.data[raw_offset]
raw_corrupt.data[raw_offset] ^= 0x80
assert_same_length(plain, raw_corrupt)
if not raw_corrupt.checksum_valid():
raise AssertionError("raw payload mutation changed superblock checksum")
raw_corrupt.save(output / "raw-data-corrupt.erofs")
evidence.append(
f"raw-data-corrupt path=/plain.bin nid={inode.nid} "
f"start_block={inode.start_block} file_offset=257 "
f"patch_offset={raw_offset} old={old_byte:#04x} "
f"new={raw_corrupt.data[raw_offset]:#04x} crc=unchanged-valid "
f"provider_length={len(plain.data)}"
)
if plain.feature_incompat & FEATURE_INCOMPAT_48BIT:
raise AssertionError("plain source unexpectedly has 48-bit feature")
if plain.root_nid == 0:
raise AssertionError("48-bit fixture needs a nonzero root nid")
blocks_lo = plain.u32(SUPER + 36)
blocks_hi = 1
total_blocks = blocks_lo | (blocks_hi << 32)
provider_length = total_blocks << plain.block_bits
large = plain.clone()
large.put_u32(SUPER + 80, large.feature_incompat | FEATURE_INCOMPAT_48BIT)
large.put_u16(SUPER + 14, blocks_hi)
large.put_u64(SUPER + 112, plain.root_nid)
save_checked(large, output / "48bit-statfs-prefix.erofs")
assert_same_length(plain, large)
evidence.append(
f"48bit-statfs-prefix feature_offset={SUPER + 80} "
f"blocks_lo_offset={SUPER + 36} blocks_lo={blocks_lo} "
f"blocks_hi_offset={SUPER + 14} blocks_hi={blocks_hi} "
f"rootnid_8b_offset={SUPER + 112} rootnid_8b={plain.root_nid} "
f"total_blocks={total_blocks} required_provider_length={provider_length} "
f"prefix_length={len(plain.data)} crc=recomputed"
)
high = large.clone()
_, high_entry = high.resolve_root_entry("/high-offset.txt")
high_inode = high.inode(high_entry.nid)
if high_inode.inode_size != 64 or high_inode.layout != 0 or high_inode.size == 0:
raise ValueError("high-offset.txt is not a nonempty extended FLAT_PLAIN file")
if high_inode.start_block_high != 0:
raise ValueError("high-offset.txt already has a high start block")
low_data_offset = high_inode.start_block_low << high.block_bits
low_data = bytes(
high.data[low_data_offset : low_data_offset + high_inode.size]
)
if low_data == bytes(high_inode.size):
raise ValueError("high-offset.txt low payload is already zero")
high.data[low_data_offset : low_data_offset + high_inode.size] = bytes(
high_inode.size
)
high.put_u16(high_inode.offset + 6, 1)
save_checked(high, output / "48bit-high-file-prefix.erofs")
assert_same_length(plain, high)
high_inode = high.inode(high_entry.nid)
high_data_offset = high_inode.start_block << high.block_bits
if high_data_offset + high_inode.size > provider_length:
raise ValueError("high-offset target exceeds declared provider")
evidence.append(
f"48bit-high-file-prefix path=/high-offset.txt nid={high_inode.nid} "
f"inode_offset={high_inode.offset} startblk_hi_offset={high_inode.offset + 6} "
f"startblk_lo={high_inode.start_block_low} startblk_hi=1 "
f"start_block={high_inode.start_block} data_offset={high_data_offset} "
f"low_decoy_offset={low_data_offset} low_decoy=zero "
f"file_size={high_inode.size} required_provider_length={provider_length} "
f"prefix_length={len(plain.data)} crc=recomputed"
)
(output / "fixture-evidence.txt").write_text(
"\n".join(evidence) + "\n", encoding="ascii"
)
write_checksums(output)
def validate_full_directory_block(
image: ErofsImage, offset: int
) -> list[DirectoryEntry]:
first_name_offset = image.u16(offset + 8)
if first_name_offset < 12 or first_name_offset % 12 != 0:
raise ValueError("invalid first name offset in full directory block")
count = first_name_offset // 12
entries = []
previous = 0
for index in range(count):
entry_offset = offset + index * 12
name_offset = image.u16(entry_offset + 8)
end_offset = (
image.u16(entry_offset + 20)
if index + 1 < count
else image.block_size
)
if (
name_offset < first_name_offset
or name_offset <= previous
or end_offset <= name_offset
or end_offset > image.block_size
):
raise ValueError("invalid full-block directory offsets")
name = bytes(
image.data[offset + name_offset : offset + end_offset]
).split(b"\0", 1)[0]
if not name:
raise ValueError("empty full-block directory name")
entries.append(
DirectoryEntry(
nid=image.u64(entry_offset),
offset=entry_offset,
name_offset=name_offset,
end_offset=end_offset,
name=name,
)
)
previous = name_offset
return entries
def make_directory_fixtures(args: argparse.Namespace) -> None:
output = args.output
if output.exists():
raise ValueError(f"output already exists: {output}")
output.mkdir(parents=True)
base = ErofsImage.load(args.base)
base.validate_superblock()
shutil.copy2(args.base, output / "namei-base.erofs")
_, wide_entry = base.resolve_root_entry("/wide")
wide = base.inode(wide_entry.nid)
if wide.layout != 2 or wide.size <= base.block_size:
raise ValueError("wide must be a multi-block FLAT_INLINE directory")
wide_block = wide.start_block << base.block_bits
entries = validate_full_directory_block(base, wide_block)
last = entries[-1]
slot_start = wide_block + last.name_offset
slot_end = wide_block + last.end_offset
padding_nul = base.data.index(0, slot_start, slot_end)
if padding_nul + 9 > wide_block + base.block_size:
raise ValueError("not enough final-name padding to patch")
evidence = [
f"wide_nid={wide.nid} inode_offset={wide.offset} block_offset={wide_block} "
f"dirents={len(entries)} final_name={last.name.decode('ascii')} "
f"padding_patch={padding_nul + 1}:{padding_nul + 9}"
]
nonzero = base.clone()
nonzero.data[padding_nul + 1 : padding_nul + 9] = b"PAD!ERO!"
save_checked(nonzero, output / "namei-padding-nonzero.erofs")
short = base.clone()
root = short.inode(short.root_nid)
if root.inode_size != 32:
raise ValueError("expected compact root inode")
short.put_u32(root.offset + 8, 8)
save_checked(short, output / "namei-corrupt-short.erofs")
evidence.append(f"short-root inode_offset={root.offset} size=8")
bad_offset = base.clone()
root = bad_offset.inode(bad_offset.root_nid)
root_entries = bad_offset.directory_entries(root)
if len(root_entries) < 2:
raise ValueError("root needs at least two entries")
bad_offset.put_u16(
root_entries[1].offset + 8, root_entries[0].name_offset
)
save_checked(bad_offset, output / "namei-corrupt-nameoff.erofs")
evidence.append(
f"bad-nameoff field_offset={root_entries[1].offset + 8} "
f"new={root_entries[0].name_offset}"
)
bad_name = base.clone()
slash_offset = slot_start + 5
if bad_name.data[slash_offset] in (0, ord("/")):
raise ValueError("chosen name byte cannot be patched")
bad_name.data[slash_offset] = ord("/")
save_checked(bad_name, output / "namei-corrupt-name.erofs")
evidence.append(f"bad-name patch_offset={slash_offset} new=0x2f")
(output / "fixture-evidence.txt").write_text(
"\n".join(evidence) + "\n", encoding="ascii"
)
write_checksums(output)
def inspect_image(args: argparse.Namespace) -> None:
image = ErofsImage.load(args.image)
print(f"image={args.image}")
print(f"provider_bytes={len(image.data)}")
print(f"magic={image.u32(SUPER):#010x} offset={SUPER}")
print(f"block_size={image.block_size} block_bits={image.block_bits}")
print(f"blocks={image.blocks} blocks_lo_offset={SUPER + 36}")
print(f"root_nid={image.root_nid}")
print(f"feature_compat={image.feature_compat:#010x} offset={SUPER + 8}")
print(f"feature_incompat={image.feature_incompat:#010x} offset={SUPER + 80}")
print(
f"checksum={image.u32(SUPER + 4):#010x} offset={SUPER + 4} "
f"calculated={image.calculated_checksum():#010x} "
f"coverage={SUPER}:{image.checksum_end} valid={image.checksum_valid()}"
)
print(
f"kernel_calculated={image.kernel_calculated_checksum():#010x} "
f"kernel_seed={KERNEL_CRC32C_SEED:#010x} "
f"kernel_coverage={SUPER + 8}:{image.checksum_end} "
f"kernel_valid={image.kernel_checksum_valid()} "
f"equivalent={image.calculated_checksum() == image.kernel_calculated_checksum()}"
)
if args.path:
_, entry = image.resolve_root_entry(args.path)
inode = image.inode(entry.nid)
print(
f"path={args.path} nid={inode.nid} inode_offset={inode.offset} "
f"i_format={inode.inode_format:#06x} layout={inode.layout} "
f"size={inode.size} start_block_low={inode.start_block_low} "
f"start_block_high={inode.start_block_high} "
f"start_block={inode.start_block}"
)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
subparsers = parser.add_subparsers(dest="command", required=True)
inspect_parser = subparsers.add_parser("inspect")
inspect_parser.add_argument("image", type=Path)
inspect_parser.add_argument("--path")
inspect_parser.set_defaults(function=inspect_image)
error_parser = subparsers.add_parser("make-error-fixtures")
error_parser.add_argument("--plain", type=Path, required=True)
error_parser.add_argument("--compressed", type=Path, required=True)
error_parser.add_argument("--deflate", type=Path, required=True)
error_parser.add_argument("--compressed-offset", type=int, required=True)
error_parser.add_argument("--compressed-length", type=int, required=True)
error_parser.add_argument("--output", type=Path, required=True)
error_parser.set_defaults(function=make_error_fixtures)
directory_parser = subparsers.add_parser("make-directory-fixtures")
directory_parser.add_argument("--base", type=Path, required=True)
directory_parser.add_argument("--output", type=Path, required=True)
directory_parser.set_defaults(function=make_directory_fixtures)
return parser.parse_args()
def main() -> None:
args = parse_args()
args.function(args)
if __name__ == "__main__":
main()