diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..ab3c419 --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,16 @@ +version: 2 + +updates: + # No pip ecosystem entry by design: Strata is standard-library-only in + # the engine and in the tests (see CONTRIBUTING.md), so there is no + # requirements.txt / pyproject.toml to keep current. Only the CI + # tooling itself -- the GitHub Actions used in the workflows -- has + # versions worth Dependabot's attention. + - package-ecosystem: github-actions + directory: "/" + schedule: + interval: weekly + groups: + actions: + patterns: + - "*" diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml new file mode 100644 index 0000000..7e1902b --- /dev/null +++ b/.github/workflows/codeql.yml @@ -0,0 +1,56 @@ +name: CodeQL + +on: + push: + branches: [main] + pull_request: + branches: [main] + schedule: + # Weekly as well as on push: CodeQL's query packs are updated + # independently of this repo, so new checks reach unchanged code. + - cron: '17 3 * * 1' + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: codeql-${{ github.ref }} + cancel-in-progress: true + +jobs: + analyze: + name: Analyze (${{ matrix.language }}) + runs-on: ubuntu-latest + permissions: + security-events: write + contents: read + actions: read + strategy: + fail-fast: false + matrix: + # engine/ parses untrusted forensic input (images, filesystems, + # registry hives, archives, ...) byte by byte, and web/ is the + # local HTTP UI that talks to it -- both are exactly the kind of + # code CodeQL is worth running against. No compiled languages + # here, so build-mode is always "none". + include: + - { language: python, build-mode: none } + - { language: javascript-typescript, build-mode: none } + steps: + - uses: actions/checkout@v4 + + - name: Initialize CodeQL + uses: github/codeql-action/init@v4 + with: + languages: ${{ matrix.language }} + build-mode: ${{ matrix.build-mode }} + queries: security-extended + paths-ignore: | + docs/** + **/*.md + + - name: Perform CodeQL analysis + uses: github/codeql-action/analyze@v4 + with: + category: '/language:${{ matrix.language }}' diff --git a/.github/workflows/fuzz.yml b/.github/workflows/fuzz.yml new file mode 100644 index 0000000..a58147c --- /dev/null +++ b/.github/workflows/fuzz.yml @@ -0,0 +1,60 @@ +name: Fuzz + +# Mutation fuzzing of the evidence parsers (tests/fuzz.py). Kept out of the +# main CI workflow on purpose: it runs against a time budget rather than to +# completion, and a finding is something to triage, not a reason to block an +# unrelated PR. Findings are uploaded as an artifact with their inputs, and +# each can be replayed locally with `python3 tests/fuzz.py --replay `. + +on: + pull_request: + paths: + - 'engine/**' + - 'tests/fuzz.py' + - 'tests/imagebuild_*.py' + schedule: + - cron: '41 4 * * 2' + workflow_dispatch: + inputs: + minutes: + description: 'Time budget in minutes' + default: '20' + target: + description: 'Target (all, fat16, ntfs, ext4, hive, ewf, ...)' + default: 'all' + +permissions: + contents: read + +concurrency: + group: fuzz-${{ github.ref }} + cancel-in-progress: true + +jobs: + fuzz: + name: Fuzz parsers + runs-on: ubuntu-latest + # A PR gets a short pass over whatever it touched; the weekly run and a + # manual dispatch get long enough to find the less shallow problems. + continue-on-error: ${{ github.event_name == 'pull_request' }} + timeout-minutes: 75 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.13' + - name: Run the fuzzer + env: + MINUTES: ${{ github.event_name == 'pull_request' && '3' || github.event.inputs.minutes || '45' }} + TARGET: ${{ github.event.inputs.target || 'all' }} + run: | + python tests/fuzz.py --minutes "$MINUTES" --target "$TARGET" \ + --seed "${{ github.run_id }}" --out fuzz-findings + - name: Upload findings + if: always() + uses: actions/upload-artifact@v4 + with: + name: fuzz-findings + path: fuzz-findings/ + if-no-files-found: ignore + retention-days: 30 diff --git a/.gitignore b/.gitignore index 19d485d..7b19843 100644 --- a/.gitignore +++ b/.gitignore @@ -25,6 +25,7 @@ /exports/ /reports/ *-report.html +/fuzz-findings/ # Python __pycache__/ diff --git a/tests/fuzz.py b/tests/fuzz.py new file mode 100644 index 0000000..079b8c8 --- /dev/null +++ b/tests/fuzz.py @@ -0,0 +1,465 @@ +"""Mutation fuzzing for the parsers that read untrusted evidence. + +Seeds are the synthetic images from tests/imagebuild_*.py. Each case takes +one seed, applies a few random mutations (bit flips, interesting integers, +truncation, duplicated spans) and drives the same entry points the app uses: +open the filesystem and walk it, open a hive and walk it, open an E01 and +read and verify it. + +What counts as a finding: + + hang a single case ran past --case-timeout + memory the worker died or raised MemoryError under --memory-mb + exception anything other than a clean error (ValueError and the engine's + own exception classes). Reported, but only fails the run with + --strict, since the server turns these into a 500 rather than + losing the session. + +Cases run in a child worker so a hang or an out-of-memory kill takes down +only the worker; the parent records the input and starts a new one. Every +finding's input is written to --out with a JSON summary, and can be replayed: + + python3 tests/fuzz.py --minutes 5 + python3 tests/fuzz.py --target ext4 --cases 2000 --seed 7 + python3 tests/fuzz.py --replay fuzz-findings/ext4-hang-1a2b3c4d.bin + +This is meant for CI (see .github/workflows/fuzz.yml). The memory cap uses +RLIMIT_AS and is only enforced on Linux. +""" + +import argparse +import hashlib +import json +import os +import queue +import random +import struct +import subprocess +import sys +import tempfile +import threading +import time +import traceback + +HERE = os.path.dirname(os.path.abspath(__file__)) +ROOT = os.path.dirname(HERE) +sys.path.insert(0, ROOT) +sys.path.insert(0, HERE) + +# Hard caps inside each case, so a finding is about the parser rather than +# about the harness asking for too much. +WALK_BUDGET = 2000 +WALK_DEPTH = 16 +READ_FILES = 40 +READ_BYTES = 1 << 16 +HIVE_KEYS = 2000 + + +# --- targets --------------------------------------------------------------- + +class BytesImage: + """ewf.RawImage stand-in over bytes.""" + + bytes_per_sector = 512 + + def __init__(self, data): + self.data = data + self.size = len(data) + self.findings = [] + + def read_at(self, offset, length): + if offset < 0 or length <= 0 or offset >= self.size: + return b"" + return self.data[offset:offset + length] + + +def _root_handle(fs): + root = getattr(fs, "root_node", None) + if root is None: + root = {"NTFS": 5, "ext4": 2, "ext3": 2, "ext2": 2, + "APFS": 2}.get(getattr(fs, "name", ""), 0) + return root + + +def run_fs(data): + from engine import filesearch + from engine.ewf import OffsetReader + from engine.fs import ntfs + src = OffsetReader(BytesImage(data), 0, len(data)) + fs = ntfs.open_fs(src) + if hasattr(fs, "info"): + fs.info() + entries = filesearch.collect(fs, _root_handle(fs), budget=WALK_BUDGET, + max_depth=WALK_DEPTH) + read = 0 + for e in entries: + if hasattr(fs, "stat"): + fs.stat(e) + if e.get("is_dir") or read >= READ_FILES: + continue + fs.read_file(e, READ_BYTES) + read += 1 + + +def run_volume(data): + from engine import volume + volume.scan(BytesImage(data)) + + +def run_hive(data): + from engine import registry + hive = registry.open_hive(data, "SOFTWARE") + if not hive: + return + seen = 0 + stack = [hive.root()] + visited = set() + while stack and seen < HIVE_KEYS: + key = stack.pop() + if not key: + continue + marker = key.get("offset", id(key)) + if marker in visited: + continue + visited.add(marker) + seen += 1 + hive.values(key) + stack.extend(hive.subkeys(key)) + hive.carve_deleted(limit=500) + + +def run_reglog(data, _pair=[]): + import imagebuild_registry + from engine import reglog + if not _pair: + _pair.extend(imagebuild_registry.build_dirty_pair()[:2]) + reglog.recover(_pair[0], [data]) + + +def run_ewf(data, _dir=[]): + from engine import ewf + if not _dir: + _dir.append(tempfile.mkdtemp(prefix="strata-fuzz-")) + path = os.path.join(_dir[0], "case.E01") + with open(path, "wb") as fh: + fh.write(data) + img = ewf.open_image(path) + try: + step = max(1, min(img.size, 1 << 20) // 64) + for off in range(0, min(img.size, 1 << 20), step): + img.read_at(off, 4096) + img.verify() + finally: + img.close() + + +def _seeds_fs(): + import imagebuild_ext4 + import imagebuild_fat + out = { + "fat12": lambda: imagebuild_fat.build_fat(12), + "fat16": lambda: imagebuild_fat.build_fat(16), + # No fat32: its seed is ~34 MB of mostly zeros, so mutations land in + # empty space. fat12 and fat16 drive the same FatFS code paths. + "exfat": imagebuild_fat.build_exfat, + "ext4": imagebuild_ext4.build_ext4, + "ext2": imagebuild_ext4.build_ext2_legacy, + } + try: + import imagebuild_ntfs + out["ntfs"] = imagebuild_ntfs.build_ntfs + except ImportError: + pass + return out + + +def targets(): + """name -> (seed builder, runner).""" + import imagebuild_ewf + import imagebuild_fat + import imagebuild_registry + out = {} + for name, build in _seeds_fs().items(): + out[name] = (build, run_fs) + out["mbr"] = (lambda: imagebuild_fat.wrap_mbr(imagebuild_fat.build_fat(16)), + run_volume) + try: + import imagebuild_ntfs + out["gpt"] = (lambda: imagebuild_ntfs.wrap_gpt( + imagebuild_ntfs.build_ntfs()), run_volume) + except ImportError: + pass + out["hive"] = (imagebuild_registry.build_hive, run_hive) + out["reglog"] = (lambda: imagebuild_registry.build_dirty_pair()[1], + run_reglog) + out["ewf"] = (lambda: imagebuild_ewf.build_e01()[0], run_ewf) + return out + + +# --- mutation -------------------------------------------------------------- + +INTERESTING = [0, 1, 0x7F, 0x80, 0xFF, 0x100, 0x7FFF, 0x8000, 0xFFFF, + 0x7FFFFFFF, 0x80000000, 0xFFFFFFFF] + + +def mutate(seed, rng): + data = bytearray(seed) + for _ in range(rng.randint(1, 8)): + if not data: + break + op = rng.random() + pos = rng.randrange(len(data)) + if op < 0.35: + data[pos] ^= 1 << rng.randrange(8) + elif op < 0.55: + data[pos] = rng.choice((0, 0xFF, 0x7F, 0x80, rng.randrange(256))) + elif op < 0.85: + width, fmt = rng.choice(((2, "= args.cases: + break + if stop_at and time.time() >= stop_at: + break + name = names[done % len(names)] + case_seed = rng.getrandbits(48) + result = w.run(name, case_seed, args.case_timeout) + done += 1 + counts[name] += 1 + if result.get("ok"): + continue + if record(args, name, case_seed, result, findings): + print("finding: %s %s %s at %s" % ( + name, result["kind"], result.get("type"), + result.get("where", "?"))) + sys.stdout.flush() + if result["kind"] in ("hang", "memory"): + w.kill() + w = Worker(args) + finally: + w.kill() + + summary = {"cases": done, "per_target": counts, + "findings": sorted(findings.values(), + key=lambda f: (f["kind"], f["target"]))} + os.makedirs(args.out, exist_ok=True) + with open(os.path.join(args.out, "summary.json"), "w") as fh: + json.dump(summary, fh, indent=2) + kinds = {} + for f in findings.values(): + kinds[f["kind"]] = kinds.get(f["kind"], 0) + 1 + print("fuzz: %d cases, %d distinct findings %s" + % (done, len(findings), json.dumps(kinds, sort_keys=True))) + fatal = kinds.get("hang", 0) + kinds.get("memory", 0) + if args.strict: + fatal += kinds.get("exception", 0) + return 1 if fatal else 0 + + +def replay(args): + name = os.path.basename(args.replay).split("-")[0] + table = targets() + if name not in table: + sys.exit("cannot tell the target from %s" % args.replay) + with open(args.replay, "rb") as fh: + data = fh.read() + table[name][1](data) + print("replay: %s returned without raising" % name) + return 0 + + +def main(): + ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) + ap.add_argument("--target", default="all") + ap.add_argument("--minutes", type=float, default=0) + ap.add_argument("--cases", type=int, default=0) + ap.add_argument("--seed", type=int, default=0) + ap.add_argument("--case-timeout", type=float, default=10) + ap.add_argument("--memory-mb", type=int, default=1024) + ap.add_argument("--out", default=os.path.join(ROOT, "fuzz-findings")) + ap.add_argument("--strict", action="store_true") + ap.add_argument("--replay") + ap.add_argument("--worker", action="store_true", help=argparse.SUPPRESS) + args = ap.parse_args() + if args.worker: + return worker(args) + if args.replay: + return replay(args) + if not args.minutes and not args.cases: + args.cases = 500 + return fuzz(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/imagebuild_ewf.py b/tests/imagebuild_ewf.py new file mode 100644 index 0000000..0ffebba --- /dev/null +++ b/tests/imagebuild_ewf.py @@ -0,0 +1,189 @@ +"""Build synthetic EWF v1 (E01) segment files and split raw sets in memory. + +Follows the libewf "Expert Witness Compression Format" specification. + +Segment file: + + 0 file header (13 bytes): "EVF\\x09\\x0d\\x0a\\xff\\x00", 0x01, + segment number (u16), 0x0000 + 13 sections, each starting with a 76-byte descriptor: + type (16, NUL padded), next section offset (u64, absolute in + this file), section size (u64, descriptor included), + 40 bytes padding, Adler-32 of the preceding 72 bytes (u32) + +Sections written, in order: + + E01: header, volume, sectors, table, table2, then "hash" + "done" for a + single segment or "next" when another segment follows + E02+: data (copy of volume), sectors, table, table2, then next/hash+done + + header zlib-compressed text: "1", "main", tab-separated keys, values + volume 1052 bytes: media type @0, chunk count @4, sectors per chunk + @8, bytes per sector @12, sector count (u64) @16, media flags + @36, compression level @52, ... Adler-32 @1048 + sectors chunk data. Uncompressed chunk = data + Adler-32 (u32 LE); + compressed chunk = zlib stream + table entry count (u32), padding (4), base offset (u64), padding (4), + Adler-32 of those 20 bytes; entries (u32: offset relative to the + base offset, MSB set = compressed); Adler-32 of the entries + hash MD5 (16), unknown (16), Adler-32 (4) + +Chunks are CHUNK_SIZE bytes (SECTORS_PER_CHUNK x 512). build_e01() stores +even-numbered chunks zlib-compressed and odd ones uncompressed; the media's +last chunk is short when the sector count is not a whole number of chunks. +""" + +import hashlib +import struct +import zlib + +EVF_SIG = b"EVF\x09\x0d\x0a\xff\x00" +SECTOR = 512 +SECTORS_PER_CHUNK = 4 +CHUNK_SIZE = SECTOR * SECTORS_PER_CHUNK +DESC = 76 + +HEADER_FIELDS = [("c", "CASE-1"), ("n", "EV-7"), ("a", "synthetic disk"), + ("e", "Examiner"), ("t", "unit test"), ("av", "7.0"), + ("ov", "Linux"), ("m", "2026 9 16 10 0 0"), + ("u", "2026 9 16 10 0 0"), ("p", "0")] + + +def adler(data): + return zlib.adler32(data) & 0xFFFFFFFF + + +def media(sectors=14): + """Deterministic media content: `sectors` sectors, each distinct.""" + out = bytearray() + for s in range(sectors): + line = ("sector %04d " % s).encode("ascii") + out += (line * (SECTOR // len(line) + 1))[:SECTOR] + return bytes(out) + + +def descriptor(type_, offset, size, next_offset=None): + raw = type_.encode("ascii").ljust(16, b"\x00") + raw += struct.pack(" bool (default: even chunks). + chunk_mutator: function(global chunk index, stored bytes) -> bytes, to + corrupt a chunk after its checksum was computed. + """ + data = media() if data is None else data + if len(data) % SECTOR: + raise ValueError("media must be a whole number of sectors") + compress = compress or (lambda i: i % 2 == 0) + chunks = [(data[i:i + CHUNK_SIZE], compress(i // CHUNK_SIZE)) + for i in range(0, len(data), CHUNK_SIZE)] + per_segment = per_segment or len(chunks) + md5 = hashlib.md5(data).digest() + segments = [] + groups = [chunks[i:i + per_segment] + for i in range(0, len(chunks), per_segment)] + base = 0 + for n, group in enumerate(groups): + mut = None + if chunk_mutator: + mut = (lambda b: lambda i, raw: chunk_mutator(b + i, raw))(base) + segments.append(segment(n + 1, group, n == len(groups) - 1, + len(chunks), len(data) // SECTOR, md5, mut)) + base += len(group) + return segments + + +def segment_names(stem, count): + """x.E01, x.E02, ... as written by EnCase for the first 99 segments.""" + return ["%s.E%02d" % (stem, i + 1) for i in range(count)] + + +def split_raw(data, piece): + """Split raw media into .001/.002/... sized `piece` bytes.""" + return [data[i:i + piece] for i in range(0, len(data), piece)] diff --git a/tests/imagebuild_ext4.py b/tests/imagebuild_ext4.py new file mode 100644 index 0000000..88fa363 --- /dev/null +++ b/tests/imagebuild_ext4.py @@ -0,0 +1,455 @@ +"""Build small synthetic ext2/3/4 filesystem images in memory. + +Nothing here is a fixture file: every image is assembled byte by byte so the +tests stay reviewable and the builders can be reused as fuzzing seeds. + +Layout of the image returned by build_ext4() (block size 1024, one group): + + block 0 boot block (zero) + block 1 superblock (byte offset 1024) + block 2 group descriptor table (32-byte descriptors) + block 3 block bitmap + block 4 inode bitmap + blocks 5-12 inode table: 32 inodes x 256 bytes + blocks 13+ data blocks, handed out in allocation order + +Superblock features: incompat FILETYPE | EXTENTS | INLINE_DATA, compat +HAS_JOURNAL, rev 1, first_ino 11, journal inode 8. Inodes (see the INO_* +constants for the numbers and CONTENT for expected file bytes): + + 2 / directory, extent-mapped (depth 0) + 8 journal jbd2 journal, 16 blocks, extent-mapped + 12 extent.txt 2 blocks in one depth-0 extent + 13 deep.bin depth-1 extent tree: index -> leaf block -> 2 extents + 14 legacy.txt block map: 12 direct pointers + 1 single-indirect block + 15 inline.txt inline data, fits within i_block (<= 60 bytes) + 16 inline_long.txt inline data, first 60 bytes in i_block, rest in the + in-inode "system.data" extended attribute + 17 link fast symlink -> "extent.txt" + 18 sub/ directory; holds nested.txt (inode 19) + 19 sub/nested.txt 1-block extent file + 20 sparse.bin extents at logical 0 and 2 with a hole at logical 1 + 21 deleted.txt unlinked (links 0, dtime set, size 0, empty extent + header); an earlier copy of its inode table block + survives in journal transaction 5 + 22 inlinedir/ inline-data directory holding "again.txt" -> 12 + +Journal (inode 8, journal-relative block numbers): + + 0 jbd2 superblock v2, blocksize 1024, maxlen 16, first 1, sequence 5 + 1 descriptor block, sequence 5, one tag for the inode-table block that + holds inode 21 (tag flags LAST_TAG, UUID follows the tag) + 2 the superseded copy of that inode-table block (inode 21 live: links 1, + size len(CONTENT["deleted.txt"]), extent -> its old data block) + 3 commit block, sequence 5 +""" + +import struct + +BLOCK_SIZE = 1024 +INODE_SIZE = 256 +INODES_PER_GROUP = 32 +BLOCKS_COUNT = 160 +INODE_TABLE = 5 +INODE_TABLE_BLOCKS = INODES_PER_GROUP * INODE_SIZE // BLOCK_SIZE +FIRST_DATA = INODE_TABLE + INODE_TABLE_BLOCKS + +S_IFDIR = 0x4000 +S_IFREG = 0x8000 +S_IFLNK = 0xA000 +FL_EXTENTS = 0x00080000 +FL_INLINE_DATA = 0x10000000 + +INCOMPAT_FILETYPE = 0x0002 +INCOMPAT_EXTENTS = 0x0040 +INCOMPAT_INLINE_DATA = 0x8000 +COMPAT_HAS_JOURNAL = 0x0004 + +JBD2_MAGIC = 0xC03B3998 + +MTIME = 1700000000 +DTIME = 1700000500 +UUID = bytes(range(16)) +LABEL = b"strata-test" + +INO_ROOT = 2 +INO_JOURNAL = 8 +INO_EXTENT = 12 +INO_DEEP = 13 +INO_LEGACY = 14 +INO_INLINE = 15 +INO_INLINE_LONG = 16 +INO_LINK = 17 +INO_SUB = 18 +INO_NESTED = 19 +INO_SPARSE = 20 +INO_DELETED = 21 +INO_INLINEDIR = 22 + +JOURNAL_BLOCKS = 16 +JOURNAL_SEQUENCE = 5 + + +def _pattern(tag, length): + """Distinct, position-dependent filler so misplaced bytes are visible.""" + out = bytearray() + n = 0 + while len(out) < length: + out += ("%s:%05d|" % (tag, n)).encode("ascii") + n += 1 + return bytes(out[:length]) + + +CONTENT = { + "extent.txt": _pattern("extent", 1500), + "deep.bin": _pattern("deep", 3000), + "legacy.txt": _pattern("legacy", 13 * BLOCK_SIZE + 100), + "inline.txt": b"small file stored inside its own inode\n", + "inline_long.txt": _pattern("inl", 100), + "sub/nested.txt": b"nested file content\n", + "sparse.bin": (_pattern("sp0", BLOCK_SIZE) + bytes(BLOCK_SIZE) + + _pattern("sp2", BLOCK_SIZE)), + "deleted.txt": b"this content was deleted but the journal remembers\n", +} +LINK_TARGET = b"extent.txt" + + +def extent_header(entries, max_entries, depth): + """ext4_extent_header: magic 0xF30A, entries, max, depth, generation.""" + return struct.pack("> 32, + physical & 0xFFFFFFFF) + + +def extent_index(logical, leaf): + """ext4_extent_idx: ei_block, ei_leaf_lo, ei_leaf_hi, unused.""" + return struct.pack("> 32, 0) + + +def extent_area(extents, depth=0): + """A 60-byte i_block holding an extent (or index) node of up to 4.""" + body = extent_header(len(extents), 4, depth) + b"".join(extents) + return body.ljust(60, b"\x00") + + +def dirent(inode, name, ftype, rec_len=None): + """ext4_dir_entry_2: inode, rec_len, name_len, file_type, name (pad 4).""" + name = name.encode("utf-8") if isinstance(name, str) else name + need = (8 + len(name) + 3) // 4 * 4 + rec_len = rec_len or need + return struct.pack("= len(raw) else raw + else: + out += dirent(ino, name, ftype) + return bytes(out) + + +def inode_bytes(mode, size, i_block=b"", flags=0, links=1, dtime=0, + xattr=b""): + """One 256-byte inode (ext4_inode) with i_extra_isize = 32. + + `xattr` is placed at byte 160 (128 + extra_isize), i.e. the in-inode + extended attribute area, and must include its 0xEA020000 magic. + """ + raw = bytearray(INODE_SIZE) + struct.pack_into("> 32) + struct.pack_into(" self.blocks: + raise ValueError("builder ran out of blocks") + return first + + def write_block(self, block, payload): + if len(payload) > BLOCK_SIZE: + raise ValueError("payload larger than a block") + off = block * BLOCK_SIZE + self.data[off:off + BLOCK_SIZE] = payload.ljust(BLOCK_SIZE, b"\x00") + + def write_blocks(self, first, payload): + for i in range(0, len(payload), BLOCK_SIZE): + self.write_block(first + i // BLOCK_SIZE, payload[i:i + BLOCK_SIZE]) + + @staticmethod + def inode_offset(num): + return INODE_TABLE * BLOCK_SIZE + (num - 1) * INODE_SIZE + + def set_inode(self, num, raw): + off = self.inode_offset(num) + self.data[off:off + INODE_SIZE] = raw + + def superblock(self, incompat, compat, journal_inum=INO_JOURNAL): + sb = bytearray(1024) + struct.pack_into("III", sb, 0, JBD2_MAGIC, 4, 0) + struct.pack_into(">IIIII", sb, 12, BLOCK_SIZE, maxlen, first, sequence, + first) + sb[48:64] = UUID + struct.pack_into(">I", sb, 64, 1) + return bytes(sb) + + +def journal_descriptor(sequence, fs_blocks): + """Descriptor block with classic 8-byte tags (no 64bit, no csum). + + Tag: t_blocknr (be32), t_checksum (be16), t_flags (be16). The first tag + carries the 16-byte UUID after it; later tags set SAME_UUID (0x2); the + last sets LAST_TAG (0x8). + """ + blk = bytearray(struct.pack(">III", JBD2_MAGIC, 1, sequence)) + for i, b in enumerate(fs_blocks): + flags = 0x2 if i else 0 + if i == len(fs_blocks) - 1: + flags |= 0x8 + blk += struct.pack(">IHH", b, 0, flags) + if i == 0: + blk += UUID + return bytes(blk) + + +def journal_commit(sequence): + return struct.pack(">III", JBD2_MAGIC, 2, sequence) + + +def build_ext4(): + """The full test filesystem described in the module docstring.""" + img = Ext4Image() + img.superblock(INCOMPAT_FILETYPE | INCOMPAT_EXTENTS | INCOMPAT_INLINE_DATA, + COMPAT_HAS_JOURNAL) + + def reg_extents(num, data): + n = (len(data) + BLOCK_SIZE - 1) // BLOCK_SIZE + first = img.alloc(n) + img.write_blocks(first, data) + img.set_inode(num, inode_bytes(S_IFREG | 0o644, len(data), + extent_area([extent(0, n, first)]), + FL_EXTENTS)) + return first + + reg_extents(INO_EXTENT, CONTENT["extent.txt"]) + + # deep.bin: depth-1 tree. Leaf block holds two extents whose physical + # blocks are deliberately not adjacent. + deep = CONTENT["deep.bin"] + leaf = img.alloc() + a = img.alloc(2) + img.alloc() # gap on disk + b = img.alloc() + img.write_blocks(a, deep[:2 * BLOCK_SIZE]) + img.write_blocks(b, deep[2 * BLOCK_SIZE:]) + leaf_node = extent_header(2, (BLOCK_SIZE - 12) // 12, 0) + \ + extent(0, 2, a) + extent(2, 1, b) + img.write_block(leaf, leaf_node) + img.set_inode(INO_DEEP, inode_bytes( + S_IFREG | 0o644, len(deep), + extent_area([extent_index(0, leaf)], depth=1), FL_EXTENTS)) + + # legacy.txt: 14 data blocks -> 12 direct + 2 via the indirect block. + legacy = CONTENT["legacy.txt"] + n = (len(legacy) + BLOCK_SIZE - 1) // BLOCK_SIZE + first = img.alloc(n) + img.write_blocks(first, legacy) + ind = img.alloc() + img.write_block(ind, struct.pack("<%dI" % (n - 12), + *range(first + 12, first + n))) + ptrs = list(range(first, first + 12)) + [ind, 0, 0] + img.set_inode(INO_LEGACY, inode_bytes(S_IFREG | 0o644, len(legacy), + struct.pack("<15I", *ptrs))) + + small = CONTENT["inline.txt"] + img.set_inode(INO_INLINE, inode_bytes(S_IFREG | 0o644, len(small), small, + FL_INLINE_DATA)) + long_ = CONTENT["inline_long.txt"] + img.set_inode(INO_INLINE_LONG, inode_bytes( + S_IFREG | 0o644, len(long_), long_[:60], FL_INLINE_DATA, + xattr=inline_data_xattr(long_[60:]))) + + img.set_inode(INO_LINK, inode_bytes(S_IFLNK | 0o777, len(LINK_TARGET), + LINK_TARGET)) + + reg_extents(INO_NESTED, CONTENT["sub/nested.txt"]) + sub = img.alloc() + img.write_block(sub, dir_block([(INO_SUB, ".", 2), (INO_ROOT, "..", 2), + (INO_NESTED, "nested.txt", 1)])) + img.set_inode(INO_SUB, inode_bytes(S_IFDIR | 0o755, BLOCK_SIZE, + extent_area([extent(0, 1, sub)]), + FL_EXTENTS, links=2)) + + sparse = CONTENT["sparse.bin"] + s0 = img.alloc() + s2 = img.alloc() + img.write_block(s0, sparse[:BLOCK_SIZE]) + img.write_block(s2, sparse[2 * BLOCK_SIZE:]) + img.set_inode(INO_SPARSE, inode_bytes( + S_IFREG | 0o644, len(sparse), + extent_area([extent(0, 1, s0), extent(2, 1, s2)]), FL_EXTENTS)) + + # Inline directory: 4-byte parent inode, then dirents filling i_block. + inline_dir = struct.pack("IIII", JBD2_MAGIC, 5, JOURNAL_SEQUENCE, 20) + return bytes(data[:at + 14]) + + +def journal_first_block(image): + """Physical block of journal block 0 (reads inode 8's single extent).""" + off = Ext4Image.inode_offset(INO_JOURNAL) + 40 + 12 + return struct.unpack_from("> 8)) & 0xFF) for i in range(length)) + + +def fill(length, text): + """Repeat ``text`` to exactly ``length`` bytes (used to paint slack).""" + reps = length // len(text) + 1 + return (text * reps)[:length] + + +def wrap_mbr(volume, start_sector=63, ptype=0x06, bootable=False): + """A disk: classic MBR at sector 0, one primary partition holding ``volume``. + + Layout: sector 0 = MBR (partition entry 1 at 0x1BE, signature 55AA); + sectors 1..start_sector-1 zero; ``volume`` from ``start_sector``; the + disk is padded to a whole number of sectors. + """ + count = (len(volume) + SECTOR - 1) // SECTOR + mbr = bytearray(SECTOR) + entry = struct.pack(" (total_sectors, reserved, num_fats, fat_size_sectors, root_entries) +# Cluster counts land squarely inside each width's range (spec section 3.5): +# FAT12: 2847 (< 4085) FAT16: 8095 (< 65525) FAT32: 66000 (>= 65525) +FAT_GEOMETRY = { + 12: (2880, 1, 2, 9, 224), + 16: (8192, 1, 2, 32, 512), + 32: (67064, 32, 2, 516, 0), +} + +HELLO_TEXT = b"Hello, FAT!\n" +LONG_NAME = "Long file name.txt" +LONG_SHORT = b"LONGFI~1TXT" +LONG_SIZE = 2 * FAT_CLUSTER + 100 +LONG_CLUSTERS = (4, 5, 9) # fragmented chain +NESTED_SIZE = 1000 +DELETED_SIZE = FAT_CLUSTER + 188 +DELETED_LFN_NAME = "Removed document.txt" +GONE_NAME = "Gone file.txt" # exactly 13 chars: one LFN part +GONE_TEXT = b"deleted, single-part LFN\n\x00\x00\x00\x00\x00" +SLACK_MARK = b"SLACK!" + +# Where each item lives. Cluster numbers are identical across the three +# widths so the tests can share expectations. +FAT_LAYOUT = { + "root_cluster_fat32": 2, + "hello": {"cluster": 3, "size": len(HELLO_TEXT)}, + "long": {"clusters": LONG_CLUSTERS, "size": LONG_SIZE}, + "subdir": {"clusters": (6, 25)}, # two-cluster directory chain + "nested": {"clusters": (26, 27), "size": NESTED_SIZE}, + "deleted": {"cluster": 12, "size": DELETED_SIZE}, # FAT entries freed + "deleted_lfn": {"cluster": 15, "size": 50}, + "cycle": {"clusters": (17, 18), "size": 5000}, # 17 -> 18 -> 17 + "gone": {"cluster": 20, "size": 30}, + "subdir_fillers": 14, +} + + +def long_content(): + return pattern(LONG_SIZE, 1) + + +def nested_content(): + return pattern(NESTED_SIZE, 2) + + +def deleted_content(): + return pattern(DELETED_SIZE, 3) + + +def deleted_lfn_content(): + return pattern(50, 4) + + +def dos_datetime(dt): + """(date, time) as packed by FAT: 7-bit year-1980, 4 month, 5 day / + 5 hour, 6 minute, 5 two-second.""" + date = ((dt.year - 1980) << 9) | (dt.month << 5) | dt.day + time_ = (dt.hour << 11) | (dt.minute << 5) | (dt.second // 2) + return date, time_ + + +HELLO_CREATED = datetime.datetime(2024, 3, 15, 10, 20, 30) # + 1.50 s +HELLO_MODIFIED = datetime.datetime(2024, 3, 15, 13, 45, 30) +HELLO_ACCESSED = datetime.date(2024, 3, 16) + + +def lfn_checksum(short11): + s = 0 + for c in short11: + s = (((s & 1) << 7) + (s >> 1) + c) & 0xFF + return s + + +def lfn_entries(name, short11, deleted=False): + """VFAT long-name entries in on-disk order (highest sequence first).""" + units = name.encode("utf-16-le") + chars = [units[i:i + 2] for i in range(0, len(units), 2)] + parts = [chars[i:i + 13] for i in range(0, len(chars), 13)] + last = parts[-1] + if len(last) < 13: + last.append(b"\x00\x00") + while len(last) < 13: + last.append(b"\xFF\xFF") + csum = lfn_checksum(short11) + out = [] + for seq, part in enumerate(parts, 1): + raw = b"".join(part) + e = bytearray(32) + e[0] = seq | (0x40 if seq == len(parts) else 0) + e[1:11] = raw[0:10] + e[11] = 0x0F + e[13] = csum + e[14:26] = raw[10:22] + e[28:32] = raw[22:26] + if deleted: + e[0] = 0xE5 + out.append(bytes(e)) + return list(reversed(out)) + + +def short_entry(name11, attr, cluster, size, created=None, modified=None, + accessed=None, tenths=0): + e = bytearray(32) + e[0:11] = name11 + e[11] = attr + if created is not None: + d, t = dos_datetime(created) + e[13] = tenths + struct.pack_into("> 16) & 0xFFFF) + if modified is not None: + d, t = dos_datetime(modified) + struct.pack_into("> 4) & 0xFF + else: + fat[i] = value & 0xFF + fat[i + 1] = (fat[i + 1] & 0xF0) | (value >> 8) + elif bits == 16: + struct.pack_into("= 65525 clusters, ~34 MB) costs about + the size of its FATs in memory. ``extents()`` lists (offset, bytes); + ``tobytes()`` materialises the whole image.""" + + def __init__(self, size): + self.size = size + self.chunks = [] + + def __setitem__(self, key, data): + start = key.start or 0 + assert 0 <= start and start + len(data) <= self.size + self.chunks.append((start, bytes(data))) + + def extents(self): + return list(self.chunks) + + def tobytes(self): + out = bytearray(self.size) + for off, data in self.chunks: + out[off:off + len(data)] = data + return bytes(out) + + +def write_sparse(path, size, extents): + """Write ``extents`` to ``path`` as a (filesystem-sparse where supported) + file of exactly ``size`` bytes.""" + with open(path, "wb") as fh: + for off, data in extents: + fh.seek(off) + fh.write(data) + fh.truncate(size) + + +def build_fat_extents(bits=16, label=b"STRATATEST "): + """``build_fat`` as (size, [(offset, bytes), ...]) without allocating + the whole volume. Later extents overwrite earlier ones.""" + img = _build_fat(bits, label) + return img.size, img.extents() + + +def build_fat(bits=16, label=b"STRATATEST "): + """The volume from ``_build_fat`` as bytes (FAT32 is ~34 MB; prefer + ``build_fat_extents`` for it).""" + return _build_fat(bits, label).tobytes() + + +def _build_fat(bits=16, label=b"STRATATEST "): + """A FAT12, FAT16 or FAT32 volume (512-byte sectors, 1 sector/cluster). + + Geometry per width is in ``FAT_GEOMETRY``; both FAT copies are written + identically. The first data sector holds cluster 2. + + Root directory (fixed region on FAT12/16, cluster 2 on FAT32): + * volume label entry ``label`` (attr 0x08) + * ``HELLO.TXT`` 12 bytes in cluster 3; its slack is painted SLACK!; + created 2024-03-15 10:20:31.50 (tenths byte 150), modified + 2024-03-15 13:45:30, accessed 2024-03-16 + * ``Long file name.txt`` (2 LFN entries + ``LONGFI~1.TXT``), + 1124 bytes over the fragmented chain 4 -> 5 -> 9 + * ``SUBDIR`` directory, chain 6 -> 25. Cluster 6 holds ``.``, + ``..`` and 14 empty files ``FILE00.DAT``..``FILE13.DAT``; cluster 25 + holds ``NESTED.BIN`` (1000 bytes, clusters 26 -> 27) + * deleted ``DELETED.TXT`` (0xE5, no LFN), 700 bytes written + contiguously to clusters 12-13, FAT entries freed; the 324 bytes + after it in cluster 13 are painted SLACK! + * deleted ``Removed document.txt`` (2 deleted LFN entries + short + entry), 50 bytes in cluster 15, FAT entry freed + * ``CYCLE.BIN`` declares 5000 bytes on a corrupt chain 17 -> 18 -> 17 + * deleted ``Gone file.txt`` (one deleted LFN entry, 13 characters so + no terminator) + short ``GONEFI~1.TXT``, 30 bytes in cluster 20 + """ + total, reserved, nfats, fat_size, root_entries = FAT_GEOMETRY[bits] + img = SparseImage(total * SECTOR) + + # Boot sector / BPB. + b = bytearray(SECTOR) + b[0:3] = b"\xEB\x3C\x90" + b[3:11] = b"MSWIN4.1" + struct.pack_into("> 1)) + byte) & 0xFFFFFFFF + return c + + +def _name_hash(name): + h = 0 + for byte in name.upper().encode("utf-16-le"): + h = (((h << 15) | (h >> 1)) + byte) & 0xFFFF + return h + + +def _set_checksum(entries): + blob = b"".join(entries) + c = 0 + for i, byte in enumerate(blob): + if i in (2, 3): + continue + c = (((c << 15) | (c >> 1)) + byte) & 0xFFFF + first = bytearray(entries[0]) + struct.pack_into("> 1)) + byte) & 0xFFFFFFFF + return c + + +def build_exfat(): + """An exFAT volume: 512-byte sectors, 1024-byte clusters, 64 clusters. + + Sectors 0-11 main boot region (boot sector, 8 extended boot sectors with + AA550000 signatures, OEM, reserved, boot checksum sector); 12-23 backup + boot region (a copy); 24 the single FAT; 32 onwards the cluster heap + (cluster 2 at sector 32). Cluster placement is in ``EXFAT_LAYOUT``. + + Clusters: 2 allocation bitmap (8 bytes); 3 up-case table (256 bytes, + checksummed); 4 -> 15 root directory (a FAT chain). Root entries: volume + label "Strata", bitmap, up-case, then + * ``Hello.txt`` 13 bytes, cluster 6, NoFatChain; slack + painted SLACK! + * ``Fragmented file.bin`` 2500 bytes on the FAT chain 7 -> 9 -> 8 + * ``Contiguous.dat`` 3000 bytes, clusters 10-12, NoFatChain, + FAT entries left zero (as a real driver does) + * ``Deleted file.txt`` InUse cleared on all entries, 100 bytes in + cluster 13 (bitmap bit clear) + * ``Timezone.txt`` created 2024-06-01 10:00:00 local with a + valid UTC offset of +12:00 + * ``Empty.txt`` size 0, no cluster + * ``Named with a rather long filename.txt`` 3 name entries, cluster 16 + * ``Subdir`` directory, NoFatChain over clusters 17-18 + (FAT zero). Cluster 17 holds ``Inner.txt`` then unused entries + (type 0x40); cluster 18 holds ``Second.txt``. + * ``Tail.txt`` its entry set straddles clusters 4 and 15 + Every entry set carries a correct SetChecksum and NameHash. + """ + L = EXFAT_LAYOUT + cs = EXFAT_CLUSTER + total_sectors = EXFAT_HEAP_OFFSET + EXFAT_CLUSTERS * (cs // SECTOR) + img = bytearray(total_sectors * SECTOR) + + def coff(n): + return EXFAT_HEAP_OFFSET * SECTOR + (n - 2) * cs + + def put(n, data): + img[coff(n):coff(n) + len(data)] = data + + boot = bytearray(12 * SECTOR) + boot[0:3] = b"\xEB\x76\x90" + boot[3:11] = b"EXFAT " + struct.pack_into("> 7)) & 0xFF) for i in range(length)) + + +def fill(length, text): + return (text * (length // len(text) + 1))[:length] + + +def big_content(): + return pattern(BIG_SIZE, 1) + + +def secret_content(): + return pattern(SECRET_SIZE, 2) + + +def sparse_content(): + """What reading the sparse file must yield: data, zeros, data.""" + return (pattern(CLUSTER, 3) + bytes(2 * CLUSTER) + pattern(CLUSTER, 4)) + + +def fixup_content(): + return pattern(FIXUP_SIZE, 5) + + +def iso(dt): + return dt.isoformat() + "Z" + + +def filetime(dt): + delta = dt - datetime.datetime(1601, 1, 1) + return (delta.days * 86400 + delta.seconds) * 10 ** 7 \ + + delta.microseconds * 10 + + +def _align8(n): + return (n + 7) & ~7 + + +def encode_runs(runs): + """NTFS mapping pairs. ``runs`` is (lcn or None for sparse, length).""" + out = bytearray() + prev = 0 + for lcn, length in runs: + lb = (length.bit_length() + 8) // 8 + lbytes = length.to_bytes(lb, "little") + if lcn is None: + out.append(lb) + out += lbytes + continue + delta = lcn - prev + prev = lcn + ob = 1 + while True: + try: + obytes = delta.to_bytes(ob, "little", signed=True) + break + except OverflowError: + ob += 1 + out.append((ob << 4) | lb) + out += lbytes + obytes + out.append(0) + return bytes(out) + + +def resident_attr(type_id, body, name="", attr_id=0, indexed=False): + name_u = name.encode("utf-16-le") + header_len = 24 + name_off = header_len + content_off = _align8(header_len + len(name_u)) + length = _align8(content_off + len(body)) + a = bytearray(length) + struct.pack_into(" 1, + 0xFFFFFFFF -> 0xFFFFFFFE) + 0x1000 hive bins data: one "hbin" whose size is a multiple of 4096. + 32-byte header ("hbin", offset, size, reserved, timestamp, spare), + then cells: int32 size (negative = allocated, positive = free), + aligned to 8 bytes. Cell offsets are relative to 0x1000. + +build_hive() produces this key tree (root name "ROOT"): + + ROOT subkeys via an lh list + Software subkeys via an lf list + Strata values: see VALUES + System subkeys via an ri list -> [li, lf] + Alpha + Beta + Ünïcode name stored as UTF-16LE (no COMP_NAME flag) + +plus a free (positive-size) nk cell "Removed" and a free vk cell +"OldValue" = REG_SZ "gone", as left behind by a deletion. + +Transaction log (new format, .LOG1/.LOG2): + + 0x000 512-byte log base block ("regf" + sequence numbers) + 0x200 log entries: "HvLE", size (multiple of 512), flags, sequence, + hive bins data size, dirty page count, hash-1 (Marvin32 of the + entry bytes from 40 to size), hash-2 (Marvin32 of the first 32 + entry bytes, hash-1 included); then (offset, size) page + references and the page bytes. +""" + +import struct + +BASE_BLOCK = 4096 +MARVIN_SEED = 0x82EF4D887A4E55C5 +MASK = 0xFFFFFFFF +FILETIME = 133000000000000000 # 2022-06-18T04:26:40Z +BIG_DATA_SEGMENT = 16344 + +KEY_HIVE_ENTRY = 0x0004 +KEY_NO_DELETE = 0x0008 +KEY_COMP_NAME = 0x0020 +VALUE_COMP_NAME = 0x0001 + +REG_SZ, REG_BINARY, REG_DWORD, REG_MULTI_SZ = 1, 3, 4, 7 + +BIG_BLOB = bytes((i * 7 + 3) & 0xFF for i in range(20000)) + +# name -> (type, raw data) +VALUES = [ + ("", REG_SZ, "default value\x00".encode("utf-16-le")), + ("Name", REG_SZ, "Strata\x00".encode("utf-16-le")), + ("Count", REG_DWORD, struct.pack("> (32 - n))) & MASK + + +def marvin32(data, seed=MARVIN_SEED): + """Marvin32 (as used by Windows for registry log entry hashes).""" + lo, hi = seed & MASK, (seed >> 32) & MASK + + def mix(lo, hi, v): + lo = (lo + v) & MASK + hi ^= lo + lo = (_rotl(lo, 20) + hi) & MASK + hi = _rotl(hi, 9) ^ lo + lo = (_rotl(lo, 27) + hi) & MASK + hi = _rotl(hi, 19) + return lo, hi + + n = len(data) // 4 * 4 + for i in range(0, n, 4): + lo, hi = mix(lo, hi, struct.unpack_from(" BIG_DATA_SEGMENT: + size, off = len(data), self.big_data(data) + else: + size, off = len(data), self.cell(data) + body = struct.pack("<2sHIIIHH", b"vk", len(raw), size, off, vtype, + VALUE_COMP_NAME if raw else 0, 0) + return self.cell(body + raw, free=free) + + def big_data(self, data): + segs = [self.cell(data[i:i + BIG_DATA_SEGMENT]) + for i in range(0, len(data), BIG_DATA_SEGMENT)] + seglist = self.cell(struct.pack("<%dI" % len(segs), *segs)) + return self.cell(struct.pack("<2sHI", b"db", len(segs), seglist)) + + def hashed_list(self, sig, offsets): + body = struct.pack("<2sH", sig, len(offsets)) + for o in offsets: + body += struct.pack("= 0 else b"" + + def setUp(self): + self.r = ewf.OffsetReader(self.Mem(), 1000, 2000, "p1") + + def test_window(self): + self.assertEqual(self.r.bytes_per_sector, 4096) + self.assertEqual(self.r.read_at(0, 10), MEDIA[1000:1010]) + self.assertEqual(self.r.read_at(1990, 100), MEDIA[2990:3000]) + self.assertEqual(self.r.read_at(2000, 1), b"") + + def test_negative_and_empty(self): + self.assertEqual(self.r.read_at(-10, 20), b"") + self.assertEqual(self.r.read_at(10, 0), b"") + + def test_read_and_seek(self): + self.r.seek(-4, io.SEEK_END) + self.assertEqual(self.r.read(), MEDIA[2996:3000]) + self.r.seek(0) + self.assertEqual(len(self.r.read()), 2000) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_fs_exfat.py b/tests/test_fs_exfat.py new file mode 100644 index 0000000..b4f54d4 --- /dev/null +++ b/tests/test_fs_exfat.py @@ -0,0 +1,322 @@ +"""Unit tests for the exFAT parser (engine.fs.exfat). + +The image comes from tests/imagebuild_fat.build_exfat() and is opened the +way the server opens evidence: ewf.open_image -> volume.scan -> OffsetReader +-> ntfs.open_fs (the filesystem dispatcher). +""" + +import os +import random +import struct +import sys +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.dirname(HERE)) +sys.path.insert(0, HERE) + +import imagebuild_fat as build # noqa: E402 +from test_fs_fat import BytesImage, ImageFiles, by_name, \ + open_first_volume # noqa: E402 +from engine import volume # noqa: E402 +from engine.ewf import OffsetReader # noqa: E402 +from engine.fs import exfat, ntfs # noqa: E402 +from engine.fs.streams import UnsupportedStream # noqa: E402 + +C = build.EXFAT_CLUSTER +L = build.EXFAT_LAYOUT + + +class Exfat(unittest.TestCase): + + @classmethod + def setUpClass(cls): + cls.files = ImageFiles() + cls.data = build.build_exfat() + cls.image = cls.files.open(cls.data) + cls.layout, cls.part, cls.fs = open_first_volume(cls.image) + cls.root = by_name(cls.fs.listdir(0)) + + @classmethod + def tearDownClass(cls): + cls.files.close() + + # -- boot record and volume metadata --------------------------------- + + def test_dispatcher_and_probe(self): + self.assertIsInstance(self.fs, exfat.ExfatFS) + self.assertEqual(volume.identify_fs(self.image, 0), "exFAT") + self.assertEqual(self.part["detected"], "exFAT") + self.assertEqual(self.part["label"], build.EXFAT_LABEL) + + def test_boot_record_fields(self): + info = self.fs.info() + self.assertEqual(info["bytes_per_sector"], 512) + self.assertEqual(info["sectors_per_cluster"], 2) + self.assertEqual(info["cluster_size"], C) + self.assertEqual(info["cluster_count"], build.EXFAT_CLUSTERS) + self.assertEqual(info["fat_count"], 1) + self.assertEqual(info["first_data_offset"], + build.EXFAT_HEAP_OFFSET * 512) + self.assertEqual(info["root_cluster"], L["root"][0]) + self.assertEqual(info["serial"], "%08X" % build.EXFAT_SERIAL) + self.assertEqual(info["revision"], "1.0") + self.assertEqual(info["percent_in_use"], 30) + self.assertFalse(info["volume_dirty"]) + self.assertEqual(info["label"], build.EXFAT_LABEL) + + def test_allocation_bitmap(self): + self.assertEqual(self.fs._bitmap_start, L["bitmap"]) + bitmap = self.fs._load_bitmap() + self.assertEqual(len(bitmap), (build.EXFAT_CLUSTERS + 7) // 8) + + def allocated(c): + return bool(bitmap[(c - 2) // 8] >> ((c - 2) % 8) & 1) + + for c in (L["bitmap"], L["upcase"], L["hello"]["cluster"], + L["contiguous"]["cluster"] + 2, L["subdir"]["cluster"] + 1): + self.assertTrue(allocated(c), c) + self.assertFalse(allocated(L["deleted"]["cluster"])) + self.assertFalse(allocated(40)) + # The bitmap's own cluster is reported as allocated space. + off = self.fs.cluster_offset(L["bitmap"]) + self.assertIn((off, off + C), self.fs.allocated_extents()) + + def test_up_case_table_is_metadata_not_a_file(self): + # The up-case entry sits between the bitmap entry and the files; + # it must neither appear as a file nor stop the root scan. + names = set(self.root) + self.assertFalse(any("upcase" in n.lower() for n in names)) + self.assertIn("Hello.txt", names) + table = self.fs.source.read_at(self.fs.cluster_offset(L["upcase"]), + 256) + self.assertEqual(table, build.upcase_table()) + + # -- directory listing ------------------------------------------------ + + def test_root_listing(self): + self.assertEqual(set(self.root), { + "Hello.txt", "Fragmented file.bin", "Contiguous.dat", + "Deleted file.txt", "Timezone.txt", "Empty.txt", + build.EXFAT_LONG_NAME, "Subdir", "Tail.txt"}) + # Directories sort first. + self.assertEqual(self.fs.listdir(0)[0]["name"], "Subdir") + self.assertTrue(self.root["Subdir"]["is_dir"]) + self.assertTrue(self.root["Subdir"]["contiguous"]) + + def test_root_directory_follows_its_fat_chain(self): + self.assertEqual(self.fs.chain(L["root"][0]), list(L["root"])) + tail = self.root["Tail.txt"] # entry set straddles clusters + self.assertEqual(self.fs.read_file(tail), b"tail") + + def test_three_part_file_name(self): + e = self.root[build.EXFAT_LONG_NAME] + self.assertEqual(e["path"], "/" + build.EXFAT_LONG_NAME) + self.assertEqual(self.fs.read_file(e), b"long name\n") + + def test_timestamps_and_attributes(self): + e = self.root["Hello.txt"] + self.assertEqual(e["modified"], "2023-11-05T08:30:44Z") + self.assertEqual(e["accessed"], "2023-11-05T08:30:44Z") + self.assertEqual(e["attributes"], ["archive"]) + + # Bug: exfat.py ignores the UTC-offset bytes (22-24) — _ts() takes a tz + # argument but never uses it — yet labels local times "Z". 10:00 at + # +12:00 is 22:00 UTC the previous day. + @unittest.expectedFailure + def test_timestamp_honours_utc_offset(self): + self.assertEqual(self.root["Timezone.txt"]["created"], + "2024-05-31T22:00:00Z") + + def test_subdirectory_first_cluster(self): + sub = self.root["Subdir"] + inner = by_name(self.fs.listdir(sub["start_cluster"], "/Subdir")) + self.assertIn("Inner.txt", inner) + self.assertEqual(inner["Inner.txt"]["path"], "/Subdir/Inner.txt") + self.assertEqual(self.fs.read_file(inner["Inner.txt"]), b"inner\n") + + # Bug: exfat.py listdir() always walks the FAT; a NoFatChain directory's + # FAT entries are zero, so only its first cluster is read and entries + # in later clusters (here Second.txt in cluster 18) are never listed. + @unittest.expectedFailure + def test_contiguous_subdirectory_second_cluster_listed(self): + sub = self.root["Subdir"] + inner = by_name(self.fs.listdir(sub["start_cluster"], "/Subdir")) + self.assertIn("Second.txt", inner) + + # -- file content ------------------------------------------------------ + + def test_fat_chain_stream(self): + e = self.root["Fragmented file.bin"] + self.assertFalse(e["contiguous"]) + want = build.exfat_fragmented_content() + self.assertEqual(self.fs.read_file(e), want) + self.assertEqual(self.fs.read_file(e, max_bytes=1500), want[:1500]) + self.assertEqual(self.fs.read_range(e, 1000, 1100), want[1000:2100]) + runs = self.fs.runs(e) + self.assertEqual([(r["cluster"], r["clusters"], r["used"]) + for r in runs], [(7, 1, C), (9, 1, C), (8, 1, 452)]) + + def test_no_fat_chain_contiguous_stream(self): + e = self.root["Contiguous.dat"] + self.assertTrue(e["contiguous"]) + # Nothing in the FAT describes this file; the extent comes from the + # stream extension alone. + c0 = L["contiguous"]["cluster"] + for c in range(c0, c0 + L["contiguous"]["count"]): + self.assertEqual(self.fs._fat_entry(c), 0) + want = build.exfat_contiguous_content() + self.assertEqual(self.fs.read_file(e), want) + self.assertEqual(self.fs.read_range(e, 2000, 2000), want[2000:]) + runs = self.fs.runs(e) + self.assertEqual(len(runs), 1) + self.assertEqual(runs[0]["offset"], self.fs.cluster_offset(c0)) + self.assertEqual(runs[0]["length"], 3 * C) + self.assertEqual(runs[0]["used"], 3000) + self.assertTrue(self.fs.stat(e)["contiguous"]) + + def test_empty_file(self): + e = self.root["Empty.txt"] + self.assertEqual(self.fs.read_file(e), b"") + self.assertEqual(self.fs.read_range(e, 0, 10), b"") + self.assertIsNone(self.fs.slack(e)) + self.assertEqual(self.fs.runs(e), []) + + def test_named_streams_are_refused(self): + with self.assertRaises(UnsupportedStream): + self.fs.read_file(self.root["Hello.txt"], stream="ads") + + def test_deleted_entry(self): + e = self.root["Deleted file.txt"] + self.assertTrue(e["deleted"]) + self.assertTrue(e["contiguous"]) + self.assertEqual(e["size"], 100) + self.assertEqual(self.fs.read_file(e), build.pattern(100, 13)) + self.assertIn("contiguous", self.fs.stat(e)["recovery"]) + # Deleted files are not allocated space. + off = self.fs.cluster_offset(L["deleted"]["cluster"]) + self.assertNotIn(off, {s for s, _ in self.fs.allocated_extents()}) + + def test_slack(self): + e = self.root["Hello.txt"] + n = len(build.EXFAT_HELLO) + slack = self.fs.slack(e) + self.assertEqual(slack, { + "offset": self.fs.cluster_offset(L["hello"]["cluster"]) + n, + "length": C - n}) + self.assertEqual(self.fs.source.read_at(slack["offset"], + slack["length"]), + build.fill(C - n, build.SLACK_MARK)) + self.assertEqual(self.fs.stat(e)["slack"], slack) + + cont = self.fs.slack(self.root["Contiguous.dat"]) + self.assertEqual(cont, { + "offset": self.fs.cluster_offset(L["contiguous"]["cluster"] + 2) + + (3000 - 2 * C), + "length": 3 * C - 3000}) + self.assertIsNone(self.fs.slack(self.root["Subdir"])) + + +class ExfatRobustness(unittest.TestCase): + + @classmethod + def setUpClass(cls): + cls.good = build.build_exfat() + + def open_bytes(self, data): + return ntfs.open_fs(OffsetReader(BytesImage(data), 0, len(data))) + + def test_truncated_to_boot_sector(self): + files = ImageFiles() + self.addCleanup(files.close) + image = files.open(self.good[:512]) + try: + fs = ntfs.open_fs(OffsetReader(image, 0, image.size)) + except ValueError: + return + self.assertEqual(fs.listdir(0), []) + + def test_truncated_mid_heap(self): + fs = self.open_bytes(self.good[:self.good.index(b"Hello, exFAT")]) + for e in fs.listdir(0): + self.assertIsInstance(fs.read_file(e), bytes) + + def test_implausible_shifts_rejected(self): + for at, value in ((108, 0), (108, 20), (109, 40)): + data = bytearray(self.good) + data[at] = value + with self.subTest(offset=at, value=value): + with self.assertRaises(ValueError): + exfat.ExfatFS(BytesImage(data)) + + def test_signature_only_is_not_exfat(self): + data = bytearray(self.good) + data[3:11] = b"EXFAX " + with self.assertRaises(ValueError): + exfat.ExfatFS(BytesImage(data)) + + def fat_offset(self, cluster): + return build.EXFAT_FAT_OFFSET * 512 + cluster * 4 + + def test_root_chain_loop_terminates(self): + data = bytearray(self.good) + struct.pack_into(" self.read_budget: + raise RuntimeError("read budget exhausted (runaway traversal)") + if offset < 0 or offset >= self.size: + return b"" + return self.data[offset:offset + length] + + +def mount(data, offset=0, **kw): + """Open a filesystem the way the server does: OffsetReader -> open_fs.""" + img = MemImage(data, **kw) + return open_fs(OffsetReader(img, offset, img.size - offset)) + + +class Ext4Fixture(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.image = build.build_ext4() + + def setUp(self): + self.fs = mount(self.image) + self.root = {e["name"]: e for e in self.fs.listdir(2, "/")} + + +class Superblock(Ext4Fixture): + def test_open_fs_detects_ext4(self): + self.assertIsInstance(self.fs, ext4.Ext4FS) + self.assertEqual(self.fs.name, "ext4") + + def test_info_reports_superblock_fields(self): + info = self.fs.info() + self.assertEqual(info["type"], "ext4") + self.assertEqual(info["label"], "strata-test") + self.assertEqual(info["uuid"], build.UUID.hex()) + self.assertEqual(info["block_size"], 1024) + self.assertEqual(info["blocks"], build.BLOCKS_COUNT) + self.assertEqual(info["inodes"], build.INODES_PER_GROUP) + self.assertEqual(info["inode_size"], 256) + self.assertEqual(info["groups"], 1) + self.assertEqual(info["first_inode"], 11) + self.assertEqual(info["last_mounted"], "/mnt") + self.assertTrue(info["clean"]) + self.assertEqual(info["journal_inode"], 8) + for feat in ("extents", "inline_data", "filetype", "has_journal"): + self.assertIn(feat, info["features"]) + + def test_ext2_without_extents_or_journal(self): + fs = mount(build.build_ext2_legacy()) + self.assertEqual(fs.name, "ext2") + entries = fs.listdir(2) + self.assertEqual([e["name"] for e in entries], ["hello.txt"]) + self.assertEqual(fs.read_file(entries[0]), b"hello from ext2\n") + self.assertFalse(fs.journal_info()["present"]) + + def test_through_open_image_on_disk(self): + tmp = tempfile.mkdtemp(prefix="strata-ext4-test-") + self.addCleanup(lambda: shutil.rmtree(tmp, ignore_errors=True)) + path = os.path.join(tmp, "disk.dd") + pad = bytes(4096) # filesystem starts at 4096 + with open(path, "wb") as fh: + fh.write(pad + self.image) + img = open_image(path) + self.addCleanup(img.close) + self.assertIsInstance(img, RawImage) + fs = open_fs(OffsetReader(img, len(pad), img.size - len(pad))) + entry = {e["name"]: e for e in fs.listdir(2)}["extent.txt"] + self.assertEqual(fs.read_file(entry), build.CONTENT["extent.txt"]) + + +class Inodes(Ext4Fixture): + def test_inode_fields(self): + ino = self.fs.inode(build.INO_EXTENT) + self.assertEqual(ino.size, 1500) + self.assertEqual(ino.uid, 1000) + self.assertEqual(ino.links, 1) + self.assertTrue(ino.uses_extents) + self.assertFalse(ino.deleted) + self.assertEqual(ino.mtime, "2023-11-14T22:13:20Z") + self.assertEqual(ino.crtime, "2023-11-14T22:13:20Z") + self.assertEqual(ino.table_offset, build.Ext4Image.inode_offset(12)) + + def test_out_of_range_inode_is_none(self): + self.assertIsNone(self.fs.inode(0)) + self.assertIsNone(self.fs.inode(build.INODES_PER_GROUP + 1)) + + +class Directories(Ext4Fixture): + def test_root_listing(self): + self.assertEqual( + sorted(self.root), + sorted(["extent.txt", "deep.bin", "legacy.txt", "inline.txt", + "inline_long.txt", "link", "sub", "sparse.bin", + "deleted.txt", "inlinedir"])) + self.assertNotIn(".", self.root) + self.assertNotIn("..", self.root) + + def test_directories_sort_first(self): + names = [e["name"] for e in self.fs.listdir(2)] + dirs = [n for n in names if self.root[n]["is_dir"]] + self.assertEqual(names[:len(dirs)], sorted(dirs)) + + def test_entry_metadata(self): + e = self.root["extent.txt"] + self.assertEqual(e["path"], "/extent.txt") + self.assertEqual(e["inode"], build.INO_EXTENT) + self.assertEqual(e["size"], 1500) + self.assertEqual(e["mode"], "-rw-r--r--") + self.assertEqual(e["type"], "file") + self.assertEqual(e["id"], "ext:12") + self.assertFalse(e["deleted"]) + self.assertFalse(e["system"]) + self.assertTrue(self.root["sub"]["is_dir"]) + self.assertEqual(self.root["sub"]["mode"], "drwxr-xr-x") + self.assertEqual(self.root["link"]["type"], "symlink") + + def test_subdirectory(self): + entries = self.fs.listdir(build.INO_SUB, "/sub") + self.assertEqual([e["name"] for e in entries], ["nested.txt"]) + self.assertEqual(entries[0]["path"], "/sub/nested.txt") + self.assertEqual(self.fs.read_file(entries[0]), + build.CONTENT["sub/nested.txt"]) + + def test_listdir_of_a_file_is_empty(self): + self.assertEqual(self.fs.listdir(build.INO_EXTENT), []) + + # Engine bug: engine/fs/ext4.py:311 _dir_entries parses an inline-data + # directory as ordinary dirents, ignoring its 4-byte parent-inode prefix. + @unittest.expectedFailure + def test_inline_directory(self): + entries = self.fs.listdir(build.INO_INLINEDIR, "/inlinedir") + self.assertEqual([(e["name"], e["inode"]) for e in entries], + [("again.txt", build.INO_EXTENT)]) + + +class FileContent(Ext4Fixture): + def read(self, name): + return self.fs.read_file(self.root[name]) + + def test_extent_depth0(self): + self.assertEqual(self.read("extent.txt"), build.CONTENT["extent.txt"]) + self.assertEqual(self.fs.stat(self.root["extent.txt"])["mapping"], + "extent tree") + + def test_extent_depth1(self): + self.assertEqual(self.read("deep.bin"), build.CONTENT["deep.bin"]) + runs = self.fs.runs(self.fs.inode(build.INO_DEEP)) + self.assertEqual([r["blocks"] for r in runs], [2, 1]) + self.assertEqual(runs[1]["block"] - runs[0]["block"], 3) + + def test_legacy_block_map_with_indirect_block(self): + self.assertEqual(self.read("legacy.txt"), build.CONTENT["legacy.txt"]) + st = self.fs.stat(self.root["legacy.txt"]) + self.assertEqual(st["mapping"], "indirect blocks") + self.assertEqual(sum(r["blocks"] for r in st["runs"]), 14) + self.assertEqual(st["slack"]["length"], 1024 - 100) + + def test_inline_data_small(self): + self.assertEqual(self.read("inline.txt"), build.CONTENT["inline.txt"]) + self.assertTrue(self.root["inline.txt"]["inline"]) + st = self.fs.stat(self.root["inline.txt"]) + self.assertEqual(st["mapping"], "inline") + + # Engine bug: engine/fs/ext4.py:296 reads inline data past 60 bytes from + # raw[128:] (i_extra_isize...) instead of the "system.data" xattr. + @unittest.expectedFailure + def test_inline_data_spilling_into_xattr(self): + self.assertEqual(self.read("inline_long.txt"), + build.CONTENT["inline_long.txt"]) + + def test_fast_symlink(self): + self.assertEqual(self.read("link"), build.LINK_TARGET) + + def test_read_range(self): + data = build.CONTENT["legacy.txt"] + self.assertEqual( + self.fs.read_range(self.root["legacy.txt"], 12000, 2000), + data[12000:14000]) + self.assertEqual( + self.fs.read_range(self.root["inline.txt"], 6, 4), + build.CONTENT["inline.txt"][6:10]) + + def test_max_bytes(self): + self.assertEqual(self.fs.read_file(self.root["deep.bin"], 10), + build.CONTENT["deep.bin"][:10]) + + def test_named_stream_unsupported(self): + with self.assertRaises(ValueError): + self.fs.read_file(self.root["extent.txt"], stream="ads") + + # Engine bug: engine/fs/ext4.py:261-271 drops ee_block (logical start), + # so holes between extents vanish and later extents shift down. + @unittest.expectedFailure + def test_sparse_extent_hole_reads_as_zeros(self): + self.assertEqual(self.read("sparse.bin"), build.CONTENT["sparse.bin"]) + + +class Journal(Ext4Fixture): + def test_journal_info(self): + info = self.fs.journal_info() + self.assertTrue(info["present"]) + self.assertEqual(info["blocks"], build.JOURNAL_BLOCKS) + self.assertEqual(info["sequence"], build.JOURNAL_SEQUENCE) + self.assertEqual(info["uuid"], build.UUID.hex()) + self.assertEqual(info["transactions"], 1) + self.assertEqual(info["committed"], 1) + self.assertEqual(info["blocks_journalled"], 1) + self.assertEqual(info["findings"], []) + + def test_deleted_inode_is_flagged(self): + e = self.root["deleted.txt"] + self.assertTrue(e["deleted"]) + self.assertEqual(e["deleted_at"], "2023-11-14T22:21:40Z") + self.assertEqual(e["size"], 0) + + def test_superseded_inode_version_found(self): + versions = self.fs.journal.inode_versions(build.INO_DELETED) + self.assertEqual(len(versions), 1) + v = versions[0] + self.assertEqual(v["sequence"], build.JOURNAL_SEQUENCE) + self.assertTrue(v["committed"]) + self.assertFalse(v["deleted"]) + self.assertEqual(v["size"], len(build.CONTENT["deleted.txt"])) + + def test_stat_reports_journal_recovery(self): + st = self.fs.stat(self.root["deleted.txt"]) + rec = st["journal_recovery"] + self.assertEqual(rec["sequence"], build.JOURNAL_SEQUENCE) + self.assertTrue(rec["committed"]) + self.assertEqual(st["recovered_size"], + len(build.CONTENT["deleted.txt"])) + self.assertIn("journal", st["recovery"]) + + def test_read_file_returns_recovered_content(self): + self.assertEqual(self.fs.read_file(self.root["deleted.txt"]), + build.CONTENT["deleted.txt"]) + + def test_no_recovery_for_inode_not_in_journal(self): + self.assertIsNone(self.fs.journal.recover(build.INO_EXTENT)) + + +class Robustness(unittest.TestCase): + def test_garbage_is_rejected(self): + with self.assertRaises(ValueError): + ext4.Ext4FS(MemImage(bytes(range(256)) * 16)) + + def test_empty_source(self): + with self.assertRaises(ValueError): + ext4.Ext4FS(MemImage(b"")) + + def test_zero_group_sizing_is_rejected(self): + data = bytearray(build.build_ext4()) + data[1024 + 32:1024 + 36] = bytes(4) # blocks_per_group + with self.assertRaises(ValueError): + ext4.Ext4FS(MemImage(bytes(data))) + + def test_truncated_after_superblock(self): + fs = ext4.Ext4FS(MemImage(build.build_ext4()[:2048])) + self.assertEqual(fs.listdir(2), []) + self.assertIsNone(fs.inode(12)) + self.assertFalse(fs.journal_info()["present"]) + + def test_truncated_inside_inode_table(self): + fs = ext4.Ext4FS(MemImage(build.build_ext4()[:7 * 1024])) + self.assertEqual(fs.listdir(2), []) + + def test_corrupt_extent_magic_gives_no_runs(self): + data = bytearray(build.build_ext4()) + off = build.Ext4Image.inode_offset(build.INO_EXTENT) + 40 + data[off:off + 2] = b"\x00\x00" + fs = mount(bytes(data)) + self.assertEqual(fs.runs(fs.inode(build.INO_EXTENT)), []) + + def test_directory_with_zero_rec_len_terminates(self): + data = bytearray(build.build_ext4()) + fs = mount(bytes(data)) + root_block = fs.runs(fs.inode(2))[0]["block"] + data[root_block * 1024 + 4:root_block * 1024 + 6] = b"\x00\x00" + self.assertEqual(mount(bytes(data)).listdir(2), []) + + # Engine bug: engine/fs/ext4.py:214-222 bounds extent-tree depth (8) but + # not fan-out or revisits; a self-referencing index node costs 84^7 reads. + @unittest.expectedFailure + def test_self_referencing_extent_index_is_bounded(self): + fs = mount(build.build_ext4_extent_loop(), read_budget=20000) + fs.runs(fs.inode(12)) + + def test_truncated_journal_is_reported_absent(self): + fs = mount(build.build_ext4_truncated_journal()) + info = fs.journal_info() + self.assertFalse(info["present"]) + self.assertTrue(info["findings"]) + self.assertIn("extent.txt", [e["name"] for e in fs.listdir(2)]) + + # Engine bug: engine/fs/ext4.py:140-151 the _Absent journal stand-in has + # no read_recovered(), which read_file calls for deleted files (line 368). + @unittest.expectedFailure + def test_deleted_file_read_with_unreadable_journal(self): + fs = mount(build.build_ext4_truncated_journal()) + entry = {e["name"]: e for e in fs.listdir(2)}["deleted.txt"] + self.assertEqual(fs.read_file(entry), b"") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_fs_fat.py b/tests/test_fs_fat.py new file mode 100644 index 0000000..7978c7a --- /dev/null +++ b/tests/test_fs_fat.py @@ -0,0 +1,411 @@ +"""Unit tests for the FAT12/16/32 parser (engine.fs.fat) and the MBR probe. + +Images come from tests/imagebuild_fat.py. They are written to a temporary +file and opened the way the server opens evidence: ewf.open_image -> +volume.scan -> OffsetReader over the partition -> ntfs.open_fs (the +filesystem dispatcher). +""" + +import os +import random +import shutil +import sys +import tempfile +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.dirname(HERE)) +sys.path.insert(0, HERE) + +import imagebuild_fat as build # noqa: E402 +from engine import volume # noqa: E402 +from engine.ewf import OffsetReader, open_image # noqa: E402 +from engine.fs import fat, ntfs # noqa: E402 + +C = build.FAT_CLUSTER +L = build.FAT_LAYOUT + + +class BytesImage: + """In-memory stand-in for ewf.RawImage (read_at + size), used only where + a test mutates an image many times and a file per mutation is wasteful.""" + + def __init__(self, data): + self.data = bytes(data) + self.size = len(self.data) + self.bytes_per_sector = 512 + + def read_at(self, offset, length): + if offset < 0 or length <= 0 or offset >= self.size: + return b"" + return self.data[offset:offset + length] + + +class ImageFiles: + """Write built images to a temp dir and open them with ewf.open_image.""" + + def __init__(self): + self.dir = tempfile.mkdtemp(prefix="strata-fs-test-") + self.images = [] + + def open(self, data, name="image.raw"): + path = os.path.join(self.dir, "%d-%s" % (len(self.images), name)) + with open(path, "wb") as fh: + fh.write(data) + img = open_image(path) + self.images.append(img) + return img + + def open_extents(self, size, extents, name="image.raw"): + """Like ``open`` for a SparseImage's extents, never holding the + whole image in memory.""" + path = os.path.join(self.dir, "%d-%s" % (len(self.images), name)) + build.write_sparse(path, size, extents) + img = open_image(path) + self.images.append(img) + return img + + def close(self): + for img in self.images: + img.close() + shutil.rmtree(self.dir, ignore_errors=True) + + +def open_first_volume(image): + """What the server does for /api/dir: scan, pick the partition, wrap it + in an OffsetReader and hand that to the dispatcher.""" + layout = volume.scan(image) + part = next(p for p in layout["partitions"] if p.get("detected")) + src = OffsetReader(image, part["offset"], part["size"], part["slot"]) + return layout, part, ntfs.open_fs(src) + + +def by_name(entries): + return {e["name"]: e for e in entries} + + +class _FatWidth: + """Width-sensitive checks, run once per FAT width by the subclasses.""" + + bits = None + + @classmethod + def setUpClass(cls): + cls.files = ImageFiles() + cls.image = cls.files.open_extents(*build.build_fat_extents(cls.bits)) + cls.layout, cls.part, cls.fs = open_first_volume(cls.image) + cls.root = by_name(cls.fs.listdir(0)) + + @classmethod + def tearDownClass(cls): + cls.files.close() + + def test_dispatcher_returns_fat_with_detected_width(self): + self.assertIsInstance(self.fs, fat.FatFS) + self.assertEqual(self.fs.bits, self.bits) + self.assertEqual(self.fs.name, "FAT%d" % self.bits) + self.assertEqual(self.fs.info()["type"], "FAT%d" % self.bits) + + def test_volume_probe_names_the_width(self): + self.assertEqual(volume.identify_fs(self.image, 0), "FAT%d" % self.bits) + self.assertEqual(self.layout["scheme"], "None (unpartitioned)") + self.assertEqual(self.part["detected"], "FAT%d" % self.bits) + self.assertEqual(self.part["label"], "STRATATEST") + + def test_bpb_fields(self): + total, reserved, nfats, fat_size, root_entries = \ + build.FAT_GEOMETRY[self.bits] + root_sectors = (root_entries * 32 + 511) // 512 + first_data = reserved + nfats * fat_size + root_sectors + info = self.fs.info() + self.assertEqual(info["bytes_per_sector"], 512) + self.assertEqual(info["sectors_per_cluster"], 1) + self.assertEqual(info["cluster_size"], C) + self.assertEqual(info["fat_count"], nfats) + self.assertEqual(info["fat_size_sectors"], fat_size) + self.assertEqual(info["first_data_offset"], first_data * 512) + self.assertEqual(info["cluster_count"], total - first_data) + self.assertEqual(info["root_cluster"], + L["root_cluster_fat32"] if self.bits == 32 else 0) + self.assertEqual(info["label"], "STRATATEST") + self.assertEqual(self.fs.total_sectors, total) + self.assertEqual(self.fs.reserved_sectors, reserved) + self.assertEqual(self.fs.root_entries, root_entries) + + def test_root_listing(self): + self.assertEqual(set(self.root), { + "HELLO.TXT", "Long file name.txt", "SUBDIR", "_ELETED.TXT", + self.root_deleted_lfn_name(), "CYCLE.BIN", build.GONE_NAME}) + # The volume label entry is metadata, not a file. + self.assertNotIn("STRATATEST", self.root) + self.assertTrue(self.root["SUBDIR"]["is_dir"]) + self.assertEqual(self.root["HELLO.TXT"]["path"], "/HELLO.TXT") + + def root_deleted_lfn_name(self): + # Deliberately taken from the listing: its exact spelling is covered + # (as a known bug) in FatDirectoryEntries. + return next(e["name"] for e in self.root.values() + if e["short_name"] == "_EMOVE~1.TXT") + + def test_single_cluster_file_content(self): + e = self.root["HELLO.TXT"] + self.assertEqual(e["start_cluster"], L["hello"]["cluster"]) + self.assertEqual(self.fs.read_file(e), build.HELLO_TEXT) + self.assertEqual(self.fs.read_file(e, max_bytes=5), b"Hello") + + def test_fragmented_chain_content_and_runs(self): + e = self.root["Long file name.txt"] + self.assertEqual(self.fs.chain(e["start_cluster"]), + list(build.LONG_CLUSTERS)) + self.assertEqual(self.fs.read_file(e), build.long_content()) + runs = self.fs.runs(e["start_cluster"], e["size"]) + self.assertEqual( + [(r["offset"], r["length"], r["used"], r["clusters"]) + for r in runs], + [(self.fs.cluster_offset(4), 2 * C, 2 * C, 2), + (self.fs.cluster_offset(9), C, 100, 1)]) + + def test_read_range_across_fragment_boundary(self): + e = self.root["Long file name.txt"] + want = build.long_content() + self.assertEqual(self.fs.read_range(e, 1000, 100), want[1000:1100]) + # Reads are clamped to the file size. + self.assertEqual(self.fs.read_range(e, 1100, 500), want[1100:]) + self.assertEqual(self.fs.read_range(e, 5000, 10), b"") + + def test_subdirectory_spanning_two_clusters(self): + sub = self.root["SUBDIR"] + self.assertEqual(self.fs.chain(sub["start_cluster"]), + list(L["subdir"]["clusters"])) + entries = by_name(self.fs.listdir(sub["start_cluster"], "/SUBDIR")) + self.assertNotIn(".", entries) + self.assertNotIn("..", entries) + self.assertEqual(len(entries), L["subdir_fillers"] + 1) + nested = entries["NESTED.BIN"] + self.assertEqual(nested["path"], "/SUBDIR/NESTED.BIN") + self.assertEqual(self.fs.read_file(nested), build.nested_content()) + + def test_cyclic_chain_is_cut_not_followed_forever(self): + e = self.root["CYCLE.BIN"] + self.assertEqual(self.fs.chain(e["start_cluster"]), + list(L["cycle"]["clusters"])) + self.assertEqual(self.fs.read_file(e), b"A" * C + b"B" * C) + + def test_allocated_extents_cover_live_files_only(self): + extents = self.fs.allocated_extents() + starts = {s for s, _ in extents} + self.assertIn(self.fs.cluster_offset(L["hello"]["cluster"]), starts) + self.assertIn(self.fs.cluster_offset(L["nested"]["clusters"][0]), + starts) + self.assertNotIn(self.fs.cluster_offset(L["deleted"]["cluster"]), + starts) + + +class Fat12(_FatWidth, unittest.TestCase): + bits = 12 + + +class Fat16(_FatWidth, unittest.TestCase): + bits = 16 + + +class Fat32(_FatWidth, unittest.TestCase): + bits = 32 + + +class FatDirectoryEntries(unittest.TestCase): + """Entry-level behaviour that does not depend on the FAT width (FAT16).""" + + @classmethod + def setUpClass(cls): + cls.files = ImageFiles() + cls.image = cls.files.open(build.build_fat(16)) + _layout, _part, cls.fs = open_first_volume(cls.image) + cls.entries = cls.fs.listdir(0) + cls.root = by_name(cls.entries) + + @classmethod + def tearDownClass(cls): + cls.files.close() + + def test_timestamps(self): + e = self.root["HELLO.TXT"] + # 10:20:30 plus a 10 ms-resolution byte of 150 -> 31 s. + self.assertEqual(e["created"], "2024-03-15T10:20:31Z") + self.assertEqual(e["modified"], "2024-03-15T13:45:30Z") + self.assertEqual(e["accessed"], "2024-03-16T00:00:00Z") + self.assertIsNone(self.root["CYCLE.BIN"]["modified"]) + + def test_attributes(self): + self.assertEqual(self.root["HELLO.TXT"]["attributes"], ["archive"]) + self.assertEqual(self.root["SUBDIR"]["attributes"], ["directory"]) + + def test_long_filename_and_short_alias(self): + e = self.root["Long file name.txt"] + self.assertEqual(e["short_name"], "LONGFI~1.TXT") + self.assertFalse(e["deleted"]) + + def test_deleted_entry_without_lfn(self): + e = self.root["_ELETED.TXT"] + self.assertTrue(e["deleted"]) + self.assertEqual(e["size"], build.DELETED_SIZE) + # The FAT chain is gone; content is read contiguously from the start. + self.assertEqual(self.fs.read_file(e), build.deleted_content()) + self.assertEqual(self.fs.read_range(e, 500, 50), + build.deleted_content()[500:550]) + self.assertIn("recovery", self.fs.stat(e)) + + def test_deleted_entry_first_character_recovered_from_lfn(self): + # Deletion overwrites only the first byte of each entry; the long + # name keeps the character the short name lost. + e = self.root[build.GONE_NAME] + self.assertTrue(e["deleted"]) + self.assertEqual(e["short_name"], "_ONEFI~1.TXT") + self.assertEqual(self.fs.read_file(e), build.GONE_TEXT) + by_short = {x["short_name"]: x for x in self.entries if x["deleted"]} + self.assertEqual(self.fs.read_file(by_short["_EMOVE~1.TXT"]), + build.deleted_lfn_content()) + + # Bug: fat.py sorts LFN parts by (seq & 0x3F), but deletion overwrites + # every sequence byte with 0xE5, so multi-part deleted names come out in + # on-disk (reverse) order: "ent.txtRemoved docum". + @unittest.expectedFailure + def test_deleted_multi_part_lfn_is_in_order(self): + names = {e["name"] for e in self.entries if e["deleted"]} + self.assertIn(build.DELETED_LFN_NAME, names) + + def test_file_slack(self): + e = self.root["HELLO.TXT"] + slack = self.fs.slack(e) + off = self.fs.cluster_offset(L["hello"]["cluster"]) + self.assertEqual(slack, {"offset": off + len(build.HELLO_TEXT), + "length": C - len(build.HELLO_TEXT)}) + got = self.fs.source.read_at(slack["offset"], slack["length"]) + self.assertEqual(got, build.fill(slack["length"], build.SLACK_MARK)) + self.assertEqual(self.fs.stat(e)["slack"], slack) + + def test_no_slack_for_directories_or_exact_fit(self): + self.assertIsNone(self.fs.slack(self.root["SUBDIR"])) + exact = dict(self.root["HELLO.TXT"], size=C) + self.assertIsNone(self.fs.slack(exact)) + + # Bug: fat.py slack() walks the (freed) FAT chain for deleted files, so it + # reports slack inside the first cluster (offset of cluster 12 + 188), + # i.e. inside the content read_file() just returned, instead of after it. + @unittest.expectedFailure + def test_deleted_multi_cluster_file_slack_follows_its_content(self): + e = self.root["_ELETED.TXT"] + tail = build.DELETED_SIZE - C + self.assertEqual(self.fs.slack(e), { + "offset": self.fs.cluster_offset(L["deleted"]["cluster"] + 1) + + tail, + "length": C - tail}) + + +class FatInMbrPartition(unittest.TestCase): + + @classmethod + def setUpClass(cls): + cls.files = ImageFiles() + + @classmethod + def tearDownClass(cls): + cls.files.close() + + def test_partition_found_and_opened(self): + image = self.files.open(build.wrap_mbr(build.build_fat(16), 63, 0x06)) + scheme, parts = volume.parse_mbr(image) + self.assertEqual(scheme, "MBR") + self.assertEqual(len(parts), 1) + self.assertEqual(parts[0]["offset"], 63 * 512) + self.assertEqual(parts[0]["type"], "FAT16") + layout, part, fs = open_first_volume(image) + self.assertEqual(layout["scheme"], "MBR") + self.assertEqual(part["detected"], "FAT16") + self.assertEqual(part["label"], "STRATATEST") + self.assertIn("HELLO.TXT", by_name(fs.listdir(0))) + self.assertEqual(fs.read_file(by_name(fs.listdir(0))["HELLO.TXT"]), + build.HELLO_TEXT) + + def test_table_type_disagreeing_with_boot_record_is_noted(self): + image = self.files.open(build.wrap_mbr(build.build_fat(12), 63, 0x83)) + part = next(p for p in volume.scan(image)["partitions"] + if p.get("slot") == "MBR 1") + self.assertEqual(part["detected"], "FAT12") + self.assertIn("FAT12", part.get("note", "")) + + +class FatRobustness(unittest.TestCase): + + @classmethod + def setUpClass(cls): + cls.good = build.build_fat(12) + + def open_bytes(self, data): + return ntfs.open_fs(OffsetReader(BytesImage(data), 0, len(data))) + + def test_random_garbage_is_not_a_filesystem(self): + rng = random.Random(1234) + junk = bytes(rng.getrandbits(8) for _ in range(64 * 1024)) + with self.assertRaises(ValueError): + self.open_bytes(junk) + with self.assertRaises(ValueError): + fat.FatFS(BytesImage(junk)) + + def test_garbage_bpb_with_boot_signature_is_rejected(self): + rng = random.Random(99) + junk = bytearray(rng.getrandbits(8) for _ in range(8192)) + junk[11:13] = b"\x00\x03" # 768 bytes per sector: implausible + junk[510:512] = b"\x55\xAA" + with self.assertRaises(ValueError): + fat.FatFS(BytesImage(junk)) + + def test_empty_and_tiny_images(self): + for data in (b"", b"\x00" * 100, self.good[:511]): + with self.assertRaises(ValueError): + self.open_bytes(data) + + def test_truncated_volume_is_refused(self): + files = ImageFiles() + self.addCleanup(files.close) + image = files.open(self.good[:4096]) + with self.assertRaises(ValueError): + ntfs.open_fs(OffsetReader(image, 0, image.size)) + + def test_listing_a_nonsense_cluster_is_empty(self): + fs = fat.FatFS(BytesImage(self.good)) + self.assertEqual(fs.listdir(0x0FFFFFF0), []) + self.assertEqual(fs.chain(1), []) + + def test_mutated_boot_sector_fails_cleanly(self): + # Every BPB byte set to 0x00 and 0xFF in turn: the parser either + # refuses the volume with ValueError or lists it without crashing. + for at in range(11, 62): + for value in (0x00, 0xFF): + data = bytearray(self.good) + data[at] = value + try: + fs = fat.FatFS(BytesImage(data)) + except ValueError: + continue + with self.subTest(offset=at, value=value): + self.assertIsInstance(fs.listdir(0), list) + + def test_mutated_root_directory_does_not_crash(self): + fs = fat.FatFS(BytesImage(self.good)) + root = (fs.reserved_sectors + fs.num_fats * fs.fat_size) * 512 + rng = random.Random(7) + for trial in range(40): + data = bytearray(self.good) + for _ in range(16): + data[root + rng.randrange(0, 12 * 32)] = rng.getrandbits(8) + fs = fat.FatFS(BytesImage(data)) + with self.subTest(trial=trial): + for e in fs.listdir(0): + if not e["is_dir"]: + fs.read_file(e, max_bytes=1 << 16) + fs.stat(e) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_fs_ntfs.py b/tests/test_fs_ntfs.py new file mode 100644 index 0000000..664c359 --- /dev/null +++ b/tests/test_fs_ntfs.py @@ -0,0 +1,413 @@ +"""Unit tests for the NTFS parser (engine.fs.ntfs), LZNT1 (engine.fs.lznt1) +and the GPT reader (engine.volume.parse_gpt). + +Images come from tests/imagebuild_ntfs.py and are opened the way the server +opens evidence: ewf.open_image -> volume.scan -> OffsetReader -> ntfs.open_fs. +""" + +import os +import random +import struct +import sys +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.dirname(HERE)) +sys.path.insert(0, HERE) + +import imagebuild_ntfs as build # noqa: E402 +from test_fs_fat import BytesImage, ImageFiles, by_name, \ + open_first_volume # noqa: E402 +from engine import volume # noqa: E402 +from engine.ewf import OffsetReader # noqa: E402 +from engine.fs import lznt1, ntfs # noqa: E402 +from engine.fs.streams import NoSuchStream # noqa: E402 + +C = build.CLUSTER +REC = build.REC + + +class NtfsOnGpt(unittest.TestCase): + + @classmethod + def setUpClass(cls): + cls.files = ImageFiles() + cls.volume = build.build_ntfs() + cls.image = cls.files.open(build.wrap_gpt(cls.volume)) + cls.layout, cls.part, cls.fs = open_first_volume(cls.image) + cls.root = by_name(cls.fs.listdir(5)) + + @classmethod + def tearDownClass(cls): + cls.files.close() + + # -- partition table and boot sector ------------------------------------ + + def test_gpt_partition_detected(self): + scheme, parts, findings = volume.parse_gpt(self.image) + self.assertEqual(scheme, "GPT") + self.assertEqual(findings, []) + self.assertEqual(len(parts), 1) + p = parts[0] + self.assertEqual(p["type"], "Microsoft Basic Data") + self.assertEqual(p["name"], "Strata data") + self.assertEqual(p["offset"], 2048 * 512) + self.assertEqual(p["size"], len(self.volume)) + self.assertEqual(self.layout["scheme"], "GPT") + self.assertEqual(self.part["detected"], "NTFS") + self.assertEqual(self.part["label"], build.VOLUME_LABEL) + + def test_damaged_primary_gpt_falls_back_to_backup(self): + disk = bytearray(build.wrap_gpt(self.volume)) + disk[2 * 512 + 60] ^= 0xFF # inside the entry name + scheme, parts, findings = volume.parse_gpt(BytesImage(disk)) + self.assertEqual(scheme, "GPT") + self.assertEqual(len(parts), 1) + self.assertEqual(parts[0]["name"], "Strata data") + text = " ".join(findings) + self.assertIn("checksum", text) + self.assertIn("backup", text) + + def test_mbr_view_is_protective(self): + scheme, parts = volume.parse_mbr(self.image) + self.assertEqual(scheme, "GPT protective") + + def test_boot_sector_fields(self): + self.assertIsInstance(self.fs, ntfs.NtfsFS) + info = self.fs.info() + self.assertEqual(info["label"], build.VOLUME_LABEL) + self.assertEqual(info["bytes_per_sector"], 512) + self.assertEqual(info["sectors_per_cluster"], 2) + self.assertEqual(info["cluster_size"], C) + self.assertEqual(info["mft_offset"], build.MFT_LCN * C) + self.assertEqual(info["mft_record_size"], build.RECORD) + self.assertEqual(info["mft_records"], build.MFT_RECORDS) + self.assertEqual(info["serial"], "%016X" % build.SERIAL) + self.assertEqual(self.fs.mftmirr_cluster, build.MFTMIRR_LCN) + + # -- records, fixups, names, timestamps --------------------------------- + + def test_root_listing(self): + for name in ("hello.txt", "big.bin", "ads.txt", "deleted.txt", "Docs", + "sparse.bin", "compressed.txt", + "Long filename document.txt", "fixup.txt", "torn.txt", + "$MFT", "$Volume"): + self.assertIn(name, self.root) + self.assertNotIn("note.txt", self.root) + self.assertEqual(self.root["hello.txt"]["path"], "/hello.txt") + self.assertTrue(self.root["Docs"]["is_dir"]) + self.assertTrue(self.root["$MFT"]["system"]) + + def test_directory_listing(self): + docs = self.root["Docs"] + inside = self.fs.listdir(docs["mft"], "/Docs") + self.assertEqual([e["name"] for e in inside], ["note.txt"]) + self.assertEqual(inside[0]["path"], "/Docs/note.txt") + self.assertEqual(self.fs.read_file(inside[0]), build.NOTE_TEXT) + + def test_fixups_restore_bytes_at_stride_end(self): + e = self.root["fixup.txt"] + self.assertTrue(e["fixup_ok"]) + self.assertEqual(self.fs.read_file(e), build.fixup_content()) + + def test_torn_record_flagged(self): + e = self.root["torn.txt"] + self.assertFalse(e["fixup_ok"]) + self.assertTrue(self.root["hello.txt"]["fixup_ok"]) + + def test_apply_fixups_directly(self): + raw = build.mft_record(40, [build.resident_attr( + 0x80, bytes(range(256)) * 3)]) + self.assertEqual(raw[510:512], build.USN) + fixed, ok = ntfs.apply_fixups(raw, 512) + self.assertTrue(ok) + self.assertNotEqual(fixed[510:512], build.USN) + torn = build.mft_record(40, [], tear=True) + self.assertFalse(ntfs.apply_fixups(torn, 512)[1]) + self.assertEqual(ntfs.apply_fixups(b"FILE", 512), (b"FILE", False)) + + def test_si_and_fn_timestamps_are_kept_apart(self): + rec = self.fs.record(REC["hello"]) + si, fn = build.SI_TIMES, build.FN_TIMES + self.assertEqual(rec.si["created"], build.iso(si["created"])) + self.assertEqual(rec.si["modified"], build.iso(si["modified"])) + self.assertEqual(rec.si["mft_modified"], build.iso(si["mft_modified"])) + self.assertEqual(rec.si["accessed"], build.iso(si["accessed"])) + name = rec.names[0] + self.assertEqual(name["fn_created"], build.iso(fn["created"])) + self.assertEqual(name["fn_modified"], build.iso(fn["modified"])) + self.assertEqual(name["fn_mft_modified"], + build.iso(fn["mft_modified"])) + self.assertEqual(name["fn_accessed"], build.iso(fn["accessed"])) + node = self.root["hello.txt"] + self.assertEqual(node["created"], build.iso(si["created"])) + self.assertEqual(node["fn_created"], build.iso(fn["created"])) + self.assertEqual(node["fn_modified"], build.iso(fn["modified"])) + + def test_win32_name_preferred_over_dos_name(self): + e = self.root["Long filename document.txt"] + names = self.fs.stat(e)["names"] + self.assertEqual({n["name"]: n["namespace"] for n in names}, + {"Long filename document.txt": 1, "LONGFI~1.TXT": 2}) + self.assertNotIn("LONGFI~1.TXT", self.root) + + def test_deleted_record(self): + e = self.root["deleted.txt"] + self.assertTrue(e["deleted"]) + self.assertFalse(self.fs.record(REC["deleted"]).in_use) + self.assertEqual(self.fs.read_file(e), build.DELETED_TEXT) + self.assertIn("recovery", self.fs.stat(e)) + live = {s for s, _ in self.fs.allocated_extents()} + self.assertIn(40 * C, live) + + # -- $DATA -------------------------------------------------------------- + + def test_resident_data(self): + e = self.root["hello.txt"] + self.assertTrue(e["resident"]) + self.assertEqual(e["size"], len(build.HELLO_TEXT)) + self.assertEqual(self.fs.read_file(e), build.HELLO_TEXT) + self.assertEqual(self.fs.read_range(e, 7, 4), b"NTFS") + data = [a for a in self.fs.stat(e)["attributes"] + if a["type"] == "$DATA"] + self.assertTrue(data[0]["resident"]) + self.assertNotIn("slack", self.fs.stat(e)) + + def test_non_resident_data_runs_and_slack(self): + e = self.root["big.bin"] + self.assertIsNone(e["resident"]) + want = build.big_content() + self.assertEqual(self.fs.read_file(e), want) + self.assertEqual(self.fs.read_range(e, 2000, 100), want[2000:2100]) + self.assertEqual(self.fs.read_file(e, max_bytes=10), want[:10]) + info = self.fs.stat(e) + self.assertEqual( + [(r["offset"], r["length"], r["used"]) for r in info["runs"]], + [(40 * C, 2 * C, 2 * C), (60 * C, C, build.BIG_SIZE - 2 * C)]) + self.assertEqual(info["slack"], {"offset": 60 * C + C - 300, + "length": 300}) + self.assertEqual(self.fs.source.read_at(60 * C + C - 300, 300), + build.fill(300, build.SLACK_MARK)) + + def test_sparse_runs_read_as_zeros(self): + e = self.root["sparse.bin"] + want = build.sparse_content() + self.assertEqual(self.fs.read_file(e), want) + self.assertEqual(self.fs.read_range(e, 1000, 2200), + want[1000:3200]) + runs = self.fs.stat(e)["runs"] + self.assertEqual([r["sparse"] for r in runs], [False, True, False]) + + def test_decode_runlist_vector(self): + # (40 x2) (+20 -> 60 x1) (sparse x3) (-50 -> 10 x1) + raw = b"\x11\x02\x28\x11\x01\x14\x01\x03\x11\x01\xCE\x00" + self.assertEqual(ntfs.decode_runlist(raw), + [(40, 2), (60, 1), (None, 3), (10, 1)]) + # The builder encodes the same list to the same bytes. + self.assertEqual(build.encode_runs( + [(40, 2), (60, 1), (None, 3), (10, 1)]), raw) + + def test_decode_runlist_stops_at_impossible_runs(self): + findings = [] + runs = ntfs.decode_runlist(b"\x11\x02\x05\x11\x40\x01\x00", 20, + findings) + self.assertEqual(runs, [(5, 2)]) + self.assertEqual(findings, [ntfs.RUN_TOO_LONG]) + findings = [] + self.assertEqual(ntfs.decode_runlist(b"\x11\x01\xF0\x00", 20, + findings), []) + self.assertEqual(findings, [ntfs.RUN_OUT_OF_VOLUME]) + self.assertEqual(ntfs.decode_runlist(b"\x44\x01"), []) + + # -- alternate data streams --------------------------------------------- + + def test_alternate_data_streams(self): + e = self.root["ads.txt"] + self.assertEqual(e["size"], len(build.ADS_MAIN)) + self.assertEqual(e["streams"], [ + {"name": "Zone.Identifier", "size": len(build.ZONE_ID)}, + {"name": "secret", "size": build.SECRET_SIZE}]) + self.assertEqual(self.fs.read_file(e), build.ADS_MAIN) + self.assertEqual(self.fs.read_file(e, stream="Zone.Identifier"), + build.ZONE_ID) + self.assertEqual(self.fs.read_file(e, stream="secret"), + build.secret_content()) + self.assertEqual(self.fs.read_range(e, 10, 5, stream="secret"), + build.secret_content()[10:15]) + streams = {s["name"]: s for s in self.fs.streams(e)} + self.assertTrue(streams[""]["default"]) + self.assertTrue(streams["Zone.Identifier"]["resident"]) + self.assertFalse(streams["secret"]["resident"]) + info = self.fs.stat(e, "secret") + self.assertEqual(info["runs"][0]["offset"], build.SECRET_LCN * C) + self.assertEqual(info["slack"]["length"], C - build.SECRET_SIZE) + + def test_missing_stream(self): + e = self.root["ads.txt"] + with self.assertRaises(NoSuchStream) as ctx: + self.fs.read_file(e, stream="nope") + self.assertEqual(ctx.exception.available, + ["Zone.Identifier", "secret"]) + self.assertTrue(self.fs.stat(e, "nope")["stream_missing"]) + + # -- compression -------------------------------------------------------- + + def test_lznt1_compressed_attribute(self): + e = self.root["compressed.txt"] + self.assertEqual(self.fs.read_file(e), build.COMPRESSED_TEXT) + self.assertEqual(self.fs.read_range(e, 600, 30), + build.COMPRESSED_TEXT[600:630]) + info = self.fs.stat(e) + self.assertTrue(info["runs_compressed"]) + self.assertEqual(self.fs.findings, []) + + # -- known engine bugs -------------------------------------------------- + + # Bug: ntfs.py record() caches the parsed record before checking + # rec.valid (lines 428-430) and returns the cache hit unchecked (418), so + # an unformatted record is None the first time and an invalid MftRecord + # (no .sequence; stat() then raises AttributeError) every time after. + @unittest.expectedFailure + def test_invalid_record_stays_invalid_on_second_lookup(self): + fs = ntfs.NtfsFS(BytesImage(self.volume)) + self.assertIsNone(fs.record(30)) + self.assertIsNone(fs.record(30)) + + +class Lznt1(unittest.TestCase): + + def test_back_reference(self): + self.assertEqual(lznt1.decompress(b"\x05\xb0\x08abc\x06\x20"), + b"abcabcabcabc") + + def test_displacement_width_grows_with_position(self): + # 17 literals, then token 0x8007 at position 17: 5 displacement bits + # and 11 length bits -> offset 17, length 10. A decoder still on the + # 4/12 split would read offset 9 instead. + body = (b"\x00ABCDEFGH" + b"\x00IJKLMNOP" + b"\x02Q" + + struct.pack(" 127 + fs = ntfs.NtfsFS(BytesImage(data)) + self.assertEqual(fs.read_file({"mft": REC["big"]}), b"") + self.assertIn(ntfs.RUN_TOO_LONG, fs.findings) + + def test_random_record_damage(self): + base = self.record_offset(REC["big"]) + rng = random.Random(8) + for trial in range(60): + data = bytearray(self.good) + for _ in range(8): + data[base + rng.randrange(0, 510)] = rng.getrandbits(8) + with self.subTest(trial=trial): + fs = ntfs.NtfsFS(BytesImage(data)) + rec = fs.record(REC["big"]) + if rec is None: + continue + for a in rec.data_attrs(): + fs.read_attr(a, max_bytes=1 << 16) + fs.stat({"mft": REC["big"]}) + + # Bug: when record 0's $DATA cannot be read, ntfs.py line 401 assumes an + # MFT of 1 << 20 clusters, not bounded by cluster_count. A 4 KB image + # with 4 KB clusters then claims 4,194,304 records, and build_tree() + # (every listdir) reads them all: ~1.8 s here, and it scales with + # cluster size — minutes for 64 KB clusters. + @unittest.expectedFailure + def test_record_count_bounded_by_volume(self): + data = bytearray(self.good[:4096]) + data[13] = 8 # 4096-byte clusters + fs = ntfs.NtfsFS(BytesImage(data)) + most = fs.cluster_count * fs.cluster_size // fs.record_size + self.assertLessEqual(fs.record_count, most) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_gate.py b/tests/test_gate.py new file mode 100644 index 0000000..5aa629d --- /dev/null +++ b/tests/test_gate.py @@ -0,0 +1,314 @@ +"""Unit tests for the HTTP request gate (engine.server): host_allowed, +origin_allowed, and the Handler._gate method that combines them with the +Sec-Fetch-Site check. + +These lock in the anti-DNS-rebinding / anti-CSRF behaviour of the local +HTTP interface: a browser page on another origin must not be able to +drive the API just because it can reach 127.0.0.1. +""" + +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from engine import server # noqa: E402 + + +class _Headers(dict): + """Minimal stand-in for http.client.HTTPMessage: case-sensitive here + (tests pass canonical casing) but exposes the .get() the gate uses.""" + + def get(self, key, default=None): + return dict.get(self, key, default) + + +class _FakeHandler: + """Just enough of Handler for _gate(): headers + a _send() that + records what was sent instead of writing to a socket.""" + + def __init__(self, headers): + self.headers = _Headers(headers) + self.sent = None + + def _send(self, code, body, ctype="application/json"): + self.sent = (code, body) + + +class _BoundStateMixin(unittest.TestCase): + """Saves/restores the module-level _BOUND dict so tests can point the + gate at whatever bind address they're exercising without leaking + state into other tests.""" + + def setUp(self): + self._saved_bound = dict(server._BOUND) + self.addCleanup(lambda: server._BOUND.update(self._saved_bound)) + + +class HostAllowedLoopbackBind(_BoundStateMixin): + """Bound to 127.0.0.1 (the default): only loopback names/addresses, + on the bound port, get through.""" + + def setUp(self): + super().setUp() + server.set_bound_address("127.0.0.1", 8722) + + def test_matching_ip_and_port_allowed(self): + self.assertTrue(server.host_allowed("127.0.0.1:8722")) + + def test_localhost_name_and_port_allowed(self): + self.assertTrue(server.host_allowed("localhost:8722")) + + def test_wrong_port_rejected(self): + self.assertFalse(server.host_allowed("127.0.0.1:9999")) + self.assertFalse(server.host_allowed("localhost:9999")) + + def test_missing_host_rejected(self): + self.assertFalse(server.host_allowed("")) + self.assertFalse(server.host_allowed(None)) + + def test_rebinding_hostname_rejected(self): + # The classic DNS-rebinding case: attacker.example resolves to + # 127.0.0.1 at request time, but the Host header still names the + # attacker's hostname, not localhost/127.0.0.1. + self.assertFalse(server.host_allowed("evil.com:8722")) + self.assertFalse(server.host_allowed("attacker.example:8722")) + + def test_host_without_a_port_matches_on_name_alone(self): + # _split_host treats a missing port as "don't care" rather than + # "must equal the bound port" -- a bare "Host: localhost" (no + # ":8722") is accepted purely because the name is a loopback + # name. Documented here so a future change to this is deliberate. + self.assertTrue(server.host_allowed("localhost")) + self.assertTrue(server.host_allowed("127.0.0.1")) + self.assertFalse(server.host_allowed("evil.com")) + + def test_ipv6_loopback_literal_allowed(self): + self.assertTrue(server.host_allowed("[::1]:8722")) + + def test_case_insensitive(self): + self.assertTrue(server.host_allowed("LOCALHOST:8722")) + + +class HostAllowedWildcardBind(_BoundStateMixin): + """Bound to 0.0.0.0 / :: (listening on every interface): loopback + names and any syntactic IP literal are accepted, but not arbitrary + hostnames -- the server can't tell in advance which of its own + interface addresses a legitimate client will use.""" + + def test_ipv4_wildcard_allows_loopback_and_ip_literals(self): + server.set_bound_address("0.0.0.0", 8722) + self.assertTrue(server.host_allowed("127.0.0.1:8722")) + self.assertTrue(server.host_allowed("localhost:8722")) + self.assertTrue(server.host_allowed("10.0.0.5:8722")) + self.assertTrue(server.host_allowed("[::1]:8722")) + self.assertTrue(server.host_allowed("[2001:db8::1]:8722")) + + def test_ipv4_wildcard_rejects_hostnames(self): + server.set_bound_address("0.0.0.0", 8722) + self.assertFalse(server.host_allowed("attacker.example:8722")) + self.assertFalse(server.host_allowed("router.lan:8722")) + + def test_ipv6_wildcard_behaves_the_same(self): + server.set_bound_address("::", 8722) + self.assertTrue(server.host_allowed("127.0.0.1:8722")) + self.assertTrue(server.host_allowed("[::1]:8722")) + self.assertFalse(server.host_allowed("attacker.example:8722")) + + def test_wildcard_still_enforces_port(self): + server.set_bound_address("0.0.0.0", 8722) + self.assertFalse(server.host_allowed("127.0.0.1:9999")) + + +class HostAllowedSpecificLanBind(_BoundStateMixin): + """Bound to one specific LAN address: only that host (any port + variant of it) is allowed -- notably, loopback stops working.""" + + def setUp(self): + super().setUp() + server.set_bound_address("192.168.1.50", 8722) + + def test_bound_address_allowed(self): + self.assertTrue(server.host_allowed("192.168.1.50:8722")) + + def test_loopback_rejected_when_bound_elsewhere(self): + self.assertFalse(server.host_allowed("127.0.0.1:8722")) + self.assertFalse(server.host_allowed("localhost:8722")) + + def test_other_ip_rejected(self): + self.assertFalse(server.host_allowed("192.168.1.51:8722")) + self.assertFalse(server.host_allowed("10.0.0.5:8722")) + + def test_wrong_port_rejected(self): + self.assertFalse(server.host_allowed("192.168.1.50:9999")) + + +class OriginAllowed(_BoundStateMixin): + + def setUp(self): + super().setUp() + server.set_bound_address("127.0.0.1", 8722) + + def test_no_origin_header_allowed(self): + # Not every legitimate request carries Origin (plain navigations, + # same-origin requests in older browsers, non-browser tooling). + self.assertTrue(server.origin_allowed("", "127.0.0.1:8722")) + self.assertTrue(server.origin_allowed(None, "127.0.0.1:8722")) + + def test_null_origin_rejected(self): + # Sandboxed iframes / some redirects send the literal string + # "null" as Origin. + self.assertFalse(server.origin_allowed("null", "127.0.0.1:8722")) + + def test_file_scheme_rejected(self): + self.assertFalse(server.origin_allowed("file://", "127.0.0.1:8722")) + + def test_extension_scheme_rejected(self): + self.assertFalse(server.origin_allowed( + "chrome-extension://abcdefghijklmnop", "127.0.0.1:8722")) + + def test_matching_origin_allowed(self): + self.assertTrue(server.origin_allowed( + "http://127.0.0.1:8722", "127.0.0.1:8722")) + + def test_matching_origin_by_name_allowed(self): + self.assertTrue(server.origin_allowed( + "http://localhost:8722", "localhost:8722")) + + def test_cross_site_origin_rejected(self): + self.assertFalse(server.origin_allowed( + "http://evil.com:8722", "127.0.0.1:8722")) + + def test_origin_host_mismatch_against_host_header_rejected(self): + # Origin says localhost but the request claims to be addressed + # to 127.0.0.1 -- inconsistent, refuse it. + self.assertFalse(server.origin_allowed( + "http://localhost:8722", "127.0.0.1:8722")) + + def test_wrong_port_in_origin_rejected(self): + self.assertFalse(server.origin_allowed( + "http://127.0.0.1:9999", "127.0.0.1:8722")) + + def test_https_scheme_also_accepted(self): + self.assertTrue(server.origin_allowed( + "https://127.0.0.1:8722", "127.0.0.1:8722")) + + def test_garbage_origin_rejected(self): + self.assertFalse(server.origin_allowed("not-a-url", "127.0.0.1:8722")) + + +class GateHostCheck(_BoundStateMixin): + """Handler._gate(): the Host header gate, checked first (421).""" + + def setUp(self): + super().setUp() + server.set_bound_address("127.0.0.1", 8722) + + def test_bad_host_returns_421(self): + fh = _FakeHandler({"Host": "evil.com:8722"}) + self.assertFalse(server.Handler._gate(fh)) + self.assertEqual(fh.sent[0], 421) + + def test_missing_host_returns_421(self): + fh = _FakeHandler({}) + self.assertFalse(server.Handler._gate(fh)) + self.assertEqual(fh.sent[0], 421) + + def test_good_host_no_other_headers_allowed(self): + fh = _FakeHandler({"Host": "127.0.0.1:8722"}) + self.assertTrue(server.Handler._gate(fh)) + self.assertIsNone(fh.sent) + + +class GateSecFetchSiteCheck(_BoundStateMixin): + """Handler._gate(): Sec-Fetch-Site is checked before Origin, and a + cross-site value is refused outright regardless of Origin.""" + + def setUp(self): + super().setUp() + server.set_bound_address("127.0.0.1", 8722) + + def test_cross_site_rejected_with_403(self): + fh = _FakeHandler({"Host": "127.0.0.1:8722", + "Sec-Fetch-Site": "cross-site"}) + self.assertFalse(server.Handler._gate(fh)) + self.assertEqual(fh.sent[0], 403) + + def test_cross_site_rejected_even_with_matching_origin(self): + # Belt and braces: even if Origin somehow matched, a browser + # that says cross-site is telling the truth about the request's + # provenance and must not be second-guessed. + fh = _FakeHandler({"Host": "127.0.0.1:8722", + "Sec-Fetch-Site": "cross-site", + "Origin": "http://127.0.0.1:8722"}) + self.assertFalse(server.Handler._gate(fh)) + self.assertEqual(fh.sent[0], 403) + + def test_same_origin_allowed(self): + fh = _FakeHandler({"Host": "127.0.0.1:8722", + "Sec-Fetch-Site": "same-origin"}) + self.assertTrue(server.Handler._gate(fh)) + + def test_same_site_allowed(self): + fh = _FakeHandler({"Host": "127.0.0.1:8722", + "Sec-Fetch-Site": "same-site"}) + self.assertTrue(server.Handler._gate(fh)) + + def test_none_value_allowed(self): + # "none" means the browser itself typed the URL / used a + # bookmark -- not driven by another page. + fh = _FakeHandler({"Host": "127.0.0.1:8722", + "Sec-Fetch-Site": "none"}) + self.assertTrue(server.Handler._gate(fh)) + + def test_absent_header_falls_through_to_origin_check(self): + # Older browsers and non-browser clients don't send + # Sec-Fetch-Site at all; the gate must not require it. + fh = _FakeHandler({"Host": "127.0.0.1:8722"}) + self.assertTrue(server.Handler._gate(fh)) + + +class GateOriginCheck(_BoundStateMixin): + """Handler._gate(): Origin is checked last, after Host and + Sec-Fetch-Site both pass.""" + + def setUp(self): + super().setUp() + server.set_bound_address("127.0.0.1", 8722) + + def test_cross_origin_rejected_with_403(self): + fh = _FakeHandler({"Host": "127.0.0.1:8722", + "Origin": "http://evil.com:8722"}) + self.assertFalse(server.Handler._gate(fh)) + self.assertEqual(fh.sent[0], 403) + + def test_matching_origin_allowed(self): + fh = _FakeHandler({"Host": "127.0.0.1:8722", + "Origin": "http://127.0.0.1:8722"}) + self.assertTrue(server.Handler._gate(fh)) + + def test_null_origin_rejected(self): + fh = _FakeHandler({"Host": "127.0.0.1:8722", "Origin": "null"}) + self.assertFalse(server.Handler._gate(fh)) + self.assertEqual(fh.sent[0], 403) + + +class HostAllowedLooseIpLiteralCheck(_BoundStateMixin): + """_is_ip_literal() is a syntactic check only (used to decide whether + a Host string "looks like" an address rather than a hostname when + bound to a wildcard address); it does not validate octet ranges. Not + exploitable on its own -- a Host header still has to reach the + process over a socket bound to that literal interface, or be one of + the fixed loopback names -- but locked in here so a tightening or + loosening of the check is a deliberate change, not a surprise.""" + + def test_out_of_range_octets_still_treated_as_ip_shaped(self): + self.assertTrue(server._is_ip_literal("999.999.999.999")) + + def test_real_hostname_not_ip_shaped(self): + self.assertFalse(server._is_ip_literal("attacker.example")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_registry.py b/tests/test_registry.py new file mode 100644 index 0000000..d865a44 --- /dev/null +++ b/tests/test_registry.py @@ -0,0 +1,369 @@ +"""Unit tests for the registry hive parser (engine.registry) and transaction +log replay (engine.reglog), fed synthetic hives from imagebuild_registry.""" + +import os +import random +import struct +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import imagebuild_registry as build # noqa: E402 +from engine import registry, reglog # noqa: E402 + + +class HiveFixture(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.data = build.build_hive() + + def setUp(self): + self.hive = registry.open_hive(self.data, "SOFTWARE") + self.assertIsNotNone(self.hive) + + def values(self, path="Software\\Strata"): + key = self.hive.open_path(path) + return {v["name"]: v for v in self.hive.values(key)} + + +class BaseBlock(HiveFixture): + def test_info(self): + info = self.hive.info() + self.assertEqual(info["type"], "regf") + self.assertEqual(info["name"], "SOFTWARE") + self.assertEqual(info["embedded_name"], "\\??\\C:\\ROOT") + self.assertEqual(info["version"], "1.5") + self.assertEqual(info["sequence"], [1, 1]) + self.assertFalse(info["dirty"]) + self.assertEqual(info["modified"], "2022-06-18T04:26:40Z") + self.assertEqual(info["findings"], []) + + def test_base_block_checksum(self): + stored = struct.unpack_from("