From 0f24c9c643a91c9722500335b6cf8eb0f6fe1a67 Mon Sep 17 00:00:00 2001 From: deadp Date: Wed, 16 Sep 2026 20:41:31 +1200 Subject: [PATCH 1/7] Add unit tests for the HTTP request gate Locks in the Host, Origin and Sec-Fetch-Site checks that keep other origins and DNS-rebinding pages from driving the local interface. Co-Authored-By: Claude Opus 5 (1M context) --- tests/test_gate.py | 314 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 314 insertions(+) create mode 100644 tests/test_gate.py 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() From 1dab6b9696333b42d60738d9673e796755d2bb33 Mon Sep 17 00:00:00 2001 From: deadp Date: Wed, 16 Sep 2026 20:41:31 +1200 Subject: [PATCH 2/7] Add CodeQL scanning and Dependabot for GitHub Actions Co-Authored-By: Claude Opus 5 (1M context) --- .github/dependabot.yml | 16 +++++++++++ .github/workflows/codeql.yml | 56 ++++++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+) create mode 100644 .github/dependabot.yml create mode 100644 .github/workflows/codeql.yml 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 }}' From c311136a3d5dbd6a89e148435b0ced53f75fb238 Mon Sep 17 00:00:00 2001 From: deadp Date: Wed, 16 Sep 2026 20:51:14 +1200 Subject: [PATCH 3/7] Add parser tests for ext4, the registry and EWF Synthetic images are built in code (tests/imagebuild_*.py) so no evidence or binary fixture is committed. Ten engine bugs found along the way are recorded as expected failures, each naming the fault: inline data and inline directories, sparse extents, extent-tree blow-up, the deleted-file path without a journal, short and truncated hives, over-broad EWF segment discovery, truncated zlib chunks, and split raw sets. Co-Authored-By: Claude Opus 5 (1M context) --- tests/imagebuild_ewf.py | 189 +++++++++++++++ tests/imagebuild_ext4.py | 455 +++++++++++++++++++++++++++++++++++ tests/imagebuild_registry.py | 255 ++++++++++++++++++++ tests/test_ewf.py | 352 +++++++++++++++++++++++++++ tests/test_fs_ext4.py | 333 +++++++++++++++++++++++++ tests/test_registry.py | 369 ++++++++++++++++++++++++++++ 6 files changed, 1953 insertions(+) create mode 100644 tests/imagebuild_ewf.py create mode 100644 tests/imagebuild_ext4.py create mode 100644 tests/imagebuild_registry.py create mode 100644 tests/test_ewf.py create mode 100644 tests/test_fs_ext4.py create mode 100644 tests/test_registry.py 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(" 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_ext4.py b/tests/test_fs_ext4.py new file mode 100644 index 0000000..c3871dd --- /dev/null +++ b/tests/test_fs_ext4.py @@ -0,0 +1,333 @@ +"""Unit tests for the ext2/3/4 reader (engine.fs.ext4) and jbd2 journal +(engine.fs.jbd2), fed synthetic images from imagebuild_ext4.""" + +import os +import shutil +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import imagebuild_ext4 as build # noqa: E402 +from engine.ewf import OffsetReader, open_image, RawImage # noqa: E402 +from engine.fs import ext4 # noqa: E402 +from engine.fs.ntfs import open_fs # noqa: E402 + + +class MemImage(object): + """The read_at/size surface an image object offers, over bytes.""" + + bytes_per_sector = 512 + + def __init__(self, data, read_budget=None): + self.data = data + self.size = len(data) + self.reads = 0 + self.read_budget = read_budget + + def read_at(self, offset, length): + self.reads += 1 + if self.read_budget is not None and self.reads > 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_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(" Date: Wed, 16 Sep 2026 21:51:53 +1200 Subject: [PATCH 4/7] Add parser tests for FAT, exFAT and NTFS Volumes are built in code (FAT32 sparsely, as it needs ~34 MB) and opened the way the server opens them. Seven engine bugs are recorded as expected failures: scrambled deleted long names and wrong slack for deleted files on FAT; an unbounded contiguous run list, truncated contiguous directories and unconverted UTC offsets on exFAT; invalid MFT records served from the cache, and a record count that ignores volume size when $MFT is unreadable. Co-Authored-By: Claude Opus 5 --- tests/imagebuild_fat.py | 687 +++++++++++++++++++++++++++++++++++++++ tests/imagebuild_ntfs.py | 545 +++++++++++++++++++++++++++++++ tests/test_fs_exfat.py | 322 ++++++++++++++++++ tests/test_fs_fat.py | 411 +++++++++++++++++++++++ tests/test_fs_ntfs.py | 413 +++++++++++++++++++++++ 5 files changed, 2378 insertions(+) create mode 100644 tests/imagebuild_fat.py create mode 100644 tests/imagebuild_ntfs.py create mode 100644 tests/test_fs_exfat.py create mode 100644 tests/test_fs_fat.py create mode 100644 tests/test_fs_ntfs.py diff --git a/tests/imagebuild_fat.py b/tests/imagebuild_fat.py new file mode 100644 index 0000000..40bf212 --- /dev/null +++ b/tests/imagebuild_fat.py @@ -0,0 +1,687 @@ +"""Synthetic FAT12/16/32 and exFAT volume builders for the parser tests. + +Everything here returns ``bytes``: a complete, minimal but valid volume (or a +disk wrapping one) built in code, because the repository carries no binary +fixtures. The layouts are fixed and described below so a test can assert +exact offsets, and so the same images can later seed a fuzzer. + +Nothing in this module imports the engine; it only encodes the on-disk +formats (Microsoft FAT specification 1.03 and the exFAT file system +specification) so a builder mistake cannot be masked by an engine mistake. +""" + +import datetime +import struct + +SECTOR = 512 + +# -------------------------------------------------------------------------- +# Shared helpers +# -------------------------------------------------------------------------- + + +def pattern(length, seed): + """Deterministic, non-repeating-looking filler so misplaced reads show.""" + return bytes(((i * 31 + seed * 7 + (i >> 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(" 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(" +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() From d782d14d0eed939a0eb33422dd5f64b398834ee2 Mon Sep 17 00:00:00 2001 From: deadp Date: Wed, 16 Sep 2026 21:51:54 +1200 Subject: [PATCH 5/7] Add mutation fuzzing for the evidence parsers tests/fuzz.py mutates the synthetic images and drives the app's own entry points in a child worker, so hangs and memory blow-ups are caught and their inputs kept for replay. It runs in its own workflow: briefly and non-blocking on PRs that touch the engine, and at length weekly or on demand. Co-Authored-By: Claude Opus 5 --- .github/workflows/fuzz.yml | 60 +++++ .gitignore | 1 + tests/fuzz.py | 467 +++++++++++++++++++++++++++++++++++++ 3 files changed, 528 insertions(+) create mode 100644 .github/workflows/fuzz.yml create mode 100644 tests/fuzz.py 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..61466bb --- /dev/null +++ b/tests/fuzz.py @@ -0,0 +1,467 @@ +"""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: + fh = getattr(img, "_fh", None) + if fh: + fh.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()) From 38a138c572c81ef9f6ab584867f329ac57080f3a Mon Sep 17 00:00:00 2001 From: deadp Date: Wed, 16 Sep 2026 21:57:05 +1200 Subject: [PATCH 6/7] Close fuzzed images through their own close() EwfImage._fh is a method, not a handle, so the harness's cleanup raised on every EWF case and swamped the real findings. Co-Authored-By: Claude Opus 5 --- tests/fuzz.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/fuzz.py b/tests/fuzz.py index 61466bb..079b8c8 100644 --- a/tests/fuzz.py +++ b/tests/fuzz.py @@ -150,9 +150,7 @@ def run_ewf(data, _dir=[]): img.read_at(off, 4096) img.verify() finally: - fh = getattr(img, "_fh", None) - if fh: - fh.close() + img.close() def _seeds_fs(): From d5250d4565362e81bfedccde93ae5f69c8759a6b Mon Sep 17 00:00:00 2001 From: deadp Date: Wed, 16 Sep 2026 22:04:03 +1200 Subject: [PATCH 7/7] Bound every size-driven read when opening EWF images Section sizes, table base offsets, section pointers and chunk geometry all come from the segment file, and were used directly as read lengths and seek offsets. One corrupt descriptor could make Strata try to read gigabytes into memory, or raise OverflowError/OSError instead of reporting the damage. Reads now go through _read_bounded, capped by the end of the segment and by what the format can hold. Metadata sections are capped at 1 MB, headers at 16 MB, table entries at 2^20, and chunks at the declared chunk size plus zlib's worst-case expansion. Implausible sector or chunk sizes fall back to the defaults. Each clip is recorded as a finding rather than dropped. A failed open also closes its segment handles, which would otherwise stay locked on Windows. Found by tests/fuzz.py; the new HostileSizes tests fail on the previous code and every recorded EWF fuzz input now opens or fails cleanly. Co-Authored-By: Claude Opus 5 --- engine/ewf.py | 124 ++++++++++++++++++++++++++++++++++------------ tests/test_ewf.py | 90 +++++++++++++++++++++++++++++++++ 2 files changed, 183 insertions(+), 31 deletions(-) diff --git a/engine/ewf.py b/engine/ewf.py index e7cd761..96f05f5 100644 --- a/engine/ewf.py +++ b/engine/ewf.py @@ -20,6 +20,16 @@ SECTION_DESC = 76 FILE_HEADER = 13 +# Every size below comes from the segment file itself, so each read is bounded +# by what the format can legitimately hold rather than by what a damaged or +# hostile descriptor claims. Real metadata sections are a few KB; EnCase's +# largest chunk is 32768 sectors. +MAX_METADATA_SECTION = 1 << 20 +MAX_HEADER_SECTION = 16 << 20 +MAX_CHUNK_SIZE = 64 << 20 +MAX_TABLE_ENTRIES = 1 << 20 +SECTOR_SIZES = (512, 1024, 2048, 4096) + class EwfError(Exception): pass @@ -93,7 +103,14 @@ def __init__(self, path, cache_chunks=64): self._cache_max = cache_chunks self._io_lock = threading.Lock() self._pos = 0 - self._parse() + self._seg_sizes = {} + try: + self._parse() + except BaseException: + # Otherwise a failed open leaves the evidence files held open, + # which on Windows keeps them locked. + self.close() + raise def _fh(self, seg): h = self._handles.get(seg) @@ -102,6 +119,24 @@ def _fh(self, seg): self._handles[seg] = h return h + def _seg_size(self, seg): + size = self._seg_sizes.get(seg) + if size is None: + size = os.fstat(self._fh(seg).fileno()).st_size + self._seg_sizes[seg] = size + return size + + def _read_bounded(self, seg, offset, size, cap): + """Read `size` bytes at `offset`, but never past the end of the + segment and never more than `cap`. Returns (data, clipped).""" + end = self._seg_size(seg) + if offset < 0 or offset >= end: + return b"", size > 0 + want = min(size, end - offset, cap) + fh = self._fh(seg) + fh.seek(offset) + return fh.read(want), want < size + def close(self): for h in self._handles.values(): h.close() @@ -135,7 +170,7 @@ def _parse(self): self.findings.append("Section chain in segment %d exceeded " "sane length; stopped." % seg_index) break - sec = self._read_section(fh, offset) + sec = self._read_section(seg_index, offset) if sec is None: break self.sections.append((seg_index, sec)) @@ -143,19 +178,19 @@ def _parse(self): if sec.type == "sectors": sectors_extent = (sec.data_start, sec.start + sec.size) elif sec.type in ("volume", "disk"): - self._parse_volume(fh, sec) + self._parse_volume(seg_index, sec) elif sec.type == "header2": - self._parse_header(fh, sec, utf16=True) + self._parse_header(seg_index, sec, utf16=True) elif sec.type == "header": if not self.header: - self._parse_header(fh, sec, utf16=False) + self._parse_header(seg_index, sec, utf16=False) elif sec.type == "table": pending_table = sec - self._parse_table(fh, sec, seg_index, sectors_extent) + self._parse_table(sec, seg_index, sectors_extent) elif sec.type == "digest": - self._parse_digest(fh, sec) + self._parse_digest(seg_index, sec) elif sec.type == "hash": - self._parse_hash(fh, sec) + self._parse_hash(seg_index, sec) if sec.type in ("next", "done"): break @@ -174,10 +209,13 @@ def _parse(self): if self.size == 0: self.size = len(self.chunks) * self.chunk_size - def _read_section(self, fh, offset): - fh.seek(offset) - raw = fh.read(SECTION_DESC) + def _read_section(self, seg, offset): + raw, _ = self._read_bounded(seg, offset, SECTION_DESC, SECTION_DESC) if len(raw) < SECTION_DESC: + if offset >= self._seg_size(seg): + self.findings.append( + "Section pointer %d is past the end of segment %d; " + "stopped." % (offset, seg)) return None type_ = raw[0:16].split(b"\x00")[0].decode("ascii", "replace") next_offset, size = struct.unpack(" 52 else 0 + if self.bytes_per_sector not in SECTOR_SIZES: + self.findings.append( + "Volume section declares %d bytes per sector; using 512." + % self.bytes_per_sector) + self.bytes_per_sector = 512 + if self.sectors_per_chunk * self.bytes_per_sector > MAX_CHUNK_SIZE: + self.findings.append( + "Volume section declares %d sectors per chunk; using 64." + % self.sectors_per_chunk) + self.sectors_per_chunk = 64 self.chunk_size = self.sectors_per_chunk * self.bytes_per_sector self._declared_chunks = chunk_count - def _parse_header(self, fh, sec, utf16): - fh.seek(sec.data_start) - blob = fh.read(sec.data_size) + def _parse_header(self, seg, sec, utf16): + blob = self._section_data(seg, sec, MAX_HEADER_SECTION) text, over = inflate_capped(blob, 1 << 20) if over: self.findings.append("Header section inflates to more than 1 MB; " @@ -237,9 +292,8 @@ def _parse_header(self, fh, sec, utf16): self.header[names.get(k, k)] = v break - def _parse_table(self, fh, sec, seg_index, sectors_extent): - fh.seek(sec.data_start) - head = fh.read(24) + def _parse_table(self, sec, seg_index, sectors_extent): + head, _ = self._read_bounded(seg_index, sec.data_start, 24, 24) if len(head) < 24: self.findings.append("Table section header truncated.") return @@ -248,11 +302,12 @@ def _parse_table(self, fh, sec, seg_index, sectors_extent): stored = struct.unpack(" 0x0FFFFFFF: + if entry_count == 0 or entry_count > MAX_TABLE_ENTRIES: self.findings.append("Table entry count %d is implausible." % entry_count) return - raw = fh.read(entry_count * 4) + raw, _ = self._read_bounded(seg_index, sec.data_start + 24, + entry_count * 4, entry_count * 4) if len(raw) < entry_count * 4: self.findings.append("Table entries truncated at %d." % sec.start) entry_count = len(raw) // 4 @@ -276,9 +331,8 @@ def _parse_table(self, fh, sec, seg_index, sectors_extent): length = self.chunk_size + 4 self.chunks.append(Chunk(seg_index, off, length, compressed)) - def _parse_digest(self, fh, sec): - fh.seek(sec.data_start) - d = fh.read(sec.data_size) + def _parse_digest(self, seg, sec): + d = self._section_data(seg, sec, MAX_METADATA_SECTION) if len(d) >= 36: md5, sha1 = d[0:16], d[16:36] if any(md5): @@ -286,9 +340,8 @@ def _parse_digest(self, fh, sec): if any(sha1): self.stored_sha1 = sha1.hex() - def _parse_hash(self, fh, sec): - fh.seek(sec.data_start) - d = fh.read(sec.data_size) + def _parse_hash(self, seg, sec): + d = self._section_data(seg, sec, MAX_METADATA_SECTION) if len(d) >= 16 and any(d[0:16]) and not self.stored_md5: self.stored_md5 = d[0:16].hex() @@ -300,10 +353,19 @@ def _chunk(self, index): if index < 0 or index >= len(self.chunks): return b"" c = self.chunks[index] + # A stored chunk is at most the chunk plus its checksum; a compressed + # one can exceed that only by zlib's small worst-case expansion. + cap = self.chunk_size + self.chunk_size // 64 + 64 with self._io_lock: - fh = self._fh(c.seg) - fh.seek(c.offset) - raw = fh.read(c.length) + raw, _ = self._read_bounded(c.seg, c.offset, c.length, cap) + # Running into the end of the segment is not itself remarkable (a + # final chunk sized from the nominal chunk size does); a length past + # what any chunk can be, or an offset outside the file, is. + if c.length > cap or c.offset >= self._seg_size(c.seg): + self.findings.append( + "Chunk %d declares %d bytes at offset %d in segment %d; " + "only %d could be read." % (index, c.length, c.offset, c.seg, + len(raw))) if c.compressed: data, over = inflate_capped(raw, self.chunk_size) if over: diff --git a/tests/test_ewf.py b/tests/test_ewf.py index 470ad7b..899d7a2 100644 --- a/tests/test_ewf.py +++ b/tests/test_ewf.py @@ -1,6 +1,7 @@ """Unit tests for image containers (engine.ewf): EWF v1 segment sets, raw images, split raw sets and OffsetReader, fed files from imagebuild_ewf.""" +import gc import hashlib import io import os @@ -8,6 +9,7 @@ import sys import tempfile import unittest +import warnings sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) @@ -289,6 +291,94 @@ def test_section_size_beyond_file(self): self.assertEqual(img.read_at(0, len(MEDIA)), MEDIA) +class HostileSizes(TempDir): + """Sizes and offsets in a segment file are untrusted: none of them may + decide how much is allocated. Found by tests/fuzz.py, where each of + these raised MemoryError, OverflowError or OSError.""" + + HUGE = 1 << 62 + + @staticmethod + def set_size(seg, type_, size): + at = Integrity.find_section(seg, type_) + struct.pack_into("