Add XPRESS9, XPRESS10, LZ4 and SCRUB support

dissect/database/ese/c_ese.py

@@ -26,6 +26,7 @@
     COMPRESS_SCRUB          = 0x4,
     COMPRESS_XPRESS9        = 0x5,
     COMPRESS_XPRESS10       = 0x6,
+    COMPRESS_LZ4            = 0x7,
 };
 
 enum JET_coltyp {

dissect/database/ese/c_ese.pyi

@@ -23,6 +23,7 @@ class _c_ese(__cs__.cstruct):
         COMPRESS_SCRUB = ...
         COMPRESS_XPRESS9 = ...
         COMPRESS_XPRESS10 = ...
+        COMPRESS_LZ4 = ...
 
     class JET_coltyp(__cs__.Enum):
         Nil = ...

dissect/database/ese/compression.py

@@ -1,31 +1,77 @@
 from __future__ import annotations
 
 import struct
+from typing import Final
 
-from dissect.util.compression import lzxpress, sevenbit
+from dissect.util.compression import lz4, lzxpress, lzxpress9, sevenbit
+from dissect.util.hash import crc32c as _crc32c_mod
+from dissect.util.hash.crc64 import crc64 as _crc64_nvme
 
 from dissect.database.ese.c_ese import COMPRESSION_SCHEME
 
+_crc32c = _crc32c_mod.crc32c
 
-def decompress(buf: bytes) -> bytes:
+# --- ESE record header sizes ---
+
+_XPRESS9_HEADER_SIZE: Final = 5
+"""Scheme byte + u32 LE plaintext CRC-32C (compression.cxx:1691)."""
+
+_XPRESS10_HEADER_SIZE: Final = 15
+"""Scheme byte + u16 LE size + u32 LE CRC-32C + u64 LE CRC-64 (compression.cxx:1940)."""
+
+_LZ4_HEADER_SIZE: Final = 3
+"""Scheme byte + u16 LE uncompressed size (compression.cxx:2083)."""
+
+_XPRESS_HEADER_SIZE: Final = 3
+"""Scheme byte + u16 LE uncompressed size (compression.cxx:1528)."""
+
+_XPRESS10_HEADER: Final = struct.Struct("<BHIQ")
+_LZ4_HEADER: Final = struct.Struct("<BH")
+
+
+def decompress(buf: bytes, *, verify: bool = True) -> bytes:
     """Decompress the given bytes according to the encoded compression scheme.
 
+    Handles all seven ESE record compression formats:
+    ``COMPRESS_7BITASCII`` (0x1), ``COMPRESS_7BITUNICODE`` (0x2),
+    ``COMPRESS_XPRESS`` (0x3), ``COMPRESS_SCRUB`` (0x4),
+    ``COMPRESS_XPRESS9`` (0x5), ``COMPRESS_XPRESS10`` (0x6),
+    and ``COMPRESS_LZ4`` (0x7).
+
     Args:
         buf: The compressed bytes to decompress.
+        verify: When True, verify integrity checks on formats that carry them:
+            decoded-size match for XPRESS, CRC-32C for XPRESS9, CRC-64 and
+            CRC-32C for XPRESS10. LZ4 has no checksum. When False, skip all
+            integrity checks for speed or corrupt-data recovery.
 
     Raises:
-        NotImplementedError: If the buffer is compressed with an unsupported compression algorithm (XPRESS9/XPRESS10).
+        ValueError: If the buffer is a SCRUB erase marker or an integrity
+            check fails (when ``verify`` is True).
     """
     identifier = buf[0] >> 3
+
     if identifier == COMPRESSION_SCHEME.COMPRESS_7BITASCII:
         return sevenbit.decompress(buf[1:])
+
     if identifier == COMPRESSION_SCHEME.COMPRESS_7BITUNICODE:
         return sevenbit.decompress(buf[1:], wide=True)
+
     if identifier == COMPRESSION_SCHEME.COMPRESS_XPRESS:
-        return lzxpress.decompress(buf[3:])
-    if identifier in (COMPRESSION_SCHEME.COMPRESS_XPRESS9, COMPRESSION_SCHEME.COMPRESS_XPRESS10):
-        raise NotImplementedError(f"Compression not yet implemented: {COMPRESSION_SCHEME(identifier)}")
-    # Not compressed
+        return _decompress_xpress(buf, verify=verify)
+
+    if identifier == COMPRESSION_SCHEME.COMPRESS_SCRUB:
+        raise ValueError("Record is a SCRUB erase marker: no plaintext is recoverable")
+
+    if identifier == COMPRESSION_SCHEME.COMPRESS_XPRESS9:
+        return _decompress_xpress9(buf, verify=verify)
+
+    if identifier == COMPRESSION_SCHEME.COMPRESS_XPRESS10:
+        return _decompress_xpress10(buf, verify=verify)
+
+    if identifier == COMPRESSION_SCHEME.COMPRESS_LZ4:
+        return _decompress_lz4(buf)
+
     return buf
 
 
@@ -36,15 +82,97 @@ def decompress_size(buf: bytes) -> int | None:
         buf: The compressed bytes to return the decompressed size of.
 
     Raises:
-        NotImplementedError: If the buffer is compressed with an unsupported compression algorithm (XPRESS9/XPRESS10).
+        ValueError: If the buffer is a SCRUB erase marker.
     """
     identifier = buf[0] >> 3
+
     if identifier == COMPRESSION_SCHEME.COMPRESS_7BITASCII:
-        return ((buf[0] & 7) + (8 * len(buf))) // 7
+        # Low 3 header bits hold the valid bit count of the final packed byte
+        # (compression.cxx:2135-2137); the rest are full 8-bit bytes.
+        return ((len(buf) - 2) * 8 + (buf[0] & 7) + 1) // 7
+
     if identifier == COMPRESSION_SCHEME.COMPRESS_7BITUNICODE:
-        return 2 * (((buf[0] & 7) + (8 * len(buf))) // 7)
+        return 2 * (((len(buf) - 2) * 8 + (buf[0] & 7) + 1) // 7)
+
     if identifier == COMPRESSION_SCHEME.COMPRESS_XPRESS:
         return struct.unpack("<H", buf[1:3])[0]
-    if identifier in (COMPRESSION_SCHEME.COMPRESS_XPRESS9, COMPRESSION_SCHEME.COMPRESS_XPRESS10):
-        raise NotImplementedError(f"Compression not yet implemented: {COMPRESSION_SCHEME(identifier)}")
+
+    if identifier == COMPRESSION_SCHEME.COMPRESS_SCRUB:
+        raise ValueError("Record is a SCRUB erase marker: no decompressed size")
+
+    if identifier == COMPRESSION_SCHEME.COMPRESS_XPRESS9:
+        return lzxpress9.decompressed_size(buf[_XPRESS9_HEADER_SIZE:])
+
+    if identifier == COMPRESSION_SCHEME.COMPRESS_XPRESS10:
+        return struct.unpack_from("<H", buf, 1)[0]
+
+    if identifier == COMPRESSION_SCHEME.COMPRESS_LZ4:
+        return struct.unpack_from("<H", buf, 1)[0]
+
     return None
+
+
+def _decompress_xpress(buf: bytes, *, verify: bool = True) -> bytes:
+    """Decompress an XPRESS (0x3) cell: 3-byte ESE header + Plain LZ77 stream.
+
+    When ``verify`` is True, checks that the decoded length matches the
+    header's declared uncompressed size (compression.cxx:2316-2322).
+    """
+    declared = struct.unpack("<H", buf[1:3])[0]
+    plaintext = lzxpress.decompress(buf[_XPRESS_HEADER_SIZE:], max_size=declared)
+    if verify and len(plaintext) != declared:
+        raise ValueError(f"XPRESS decoded {len(plaintext)} bytes but header declares {declared}")
+    return plaintext
+
+
+def _decompress_xpress9(buf: bytes, *, verify: bool = True) -> bytes:
+    """Decompress an XPRESS9 (0x5) cell: 5-byte ESE header + XPRESS9 blocks.
+
+    When ``verify`` is True, checks the CRC-32C of the plaintext against
+    the value stored in the header (compression.cxx:2461-2467).
+    """
+    stored_crc = struct.unpack_from("<I", buf, 1)[0]
+    plaintext = lzxpress9.decompress(buf[_XPRESS9_HEADER_SIZE:])
+    if verify:
+        actual_crc = _crc32c(plaintext)
+        if actual_crc != stored_crc:
+            raise ValueError(f"XPRESS9 plaintext CRC-32C mismatch: 0x{stored_crc:08x} vs 0x{actual_crc:08x}")
+    return plaintext
+
+
+def _decompress_xpress10(buf: bytes, *, verify: bool = True) -> bytes:
+    """Decompress an XPRESS10 (0x6) cell: 15-byte ESE header + LZ4 block.
+
+    When ``verify`` is True, checks the CRC-64/NVME of the compressed payload
+    and the CRC-32C of the plaintext (compression.cxx:2513-2530).
+    """
+    _, size, stored_crc32, stored_crc64 = _XPRESS10_HEADER.unpack_from(buf)
+    payload = buf[_XPRESS10_HEADER_SIZE:]
+
+    if verify:
+        actual_crc64 = _crc64_nvme(payload)
+        if actual_crc64 != stored_crc64:
+            raise ValueError(f"XPRESS10 payload CRC-64 mismatch: 0x{stored_crc64:016x} vs 0x{actual_crc64:016x}")
+
+    plaintext = lz4.decompress(payload, size)
+    if isinstance(plaintext, tuple):
+        plaintext = plaintext[0]
+
+    if verify:
+        actual_crc32 = _crc32c(plaintext)
+        if actual_crc32 != stored_crc32:
+            raise ValueError(f"XPRESS10 plaintext CRC-32C mismatch: 0x{stored_crc32:08x} vs 0x{actual_crc32:08x}")
+
+    return plaintext
+
+
+def _decompress_lz4(buf: bytes) -> bytes:
+    """Decompress an LZ4 (0x7) cell: 3-byte ESE header + LZ4 block.
+
+    LZ4 carries no checksum, so there is nothing to verify.
+    """
+    _, size = _LZ4_HEADER.unpack_from(buf)
+    result = lz4.decompress(buf[_LZ4_HEADER_SIZE:], size)
+    if isinstance(result, tuple):
+        result = result[0]
+    return result

tests/ese/test_compression.py

@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+import struct
+
+import pytest
+
+from dissect.database.ese.compression import decompress, decompress_size
+
+# Real ESE cells captured from esent.dll (Server 2022 Build 20348 for 7-bit/XPRESS,
+# Win11 Build 26100 for LZ4/XPRESS10) and the MIT ESE C reference encoder (XPRESS9).
+# Each cell decompresses to the paired plaintext byte-for-byte.
+FOX = b"the quick brown fox jumps over the lazy dog. " * 8
+PATTERN = bytes(0x41 + ((i + 1) % 26) for i in range(4096))
+
+CELLS = [
+    pytest.param("0f54741914afa7c76b9058febebb41", b"The quick brown ", id="7bit_ascii"),
+    pytest.param("1700000000000000", b"\x00" * 16, id="7bit_unicode"),
+    pytest.param("180004ffffff3f000007000ffffb03", b"\x00" * 1024, id="xpress"),
+    pytest.param(
+        "28f83ea8df2ad7864e68010000d00200001b00060000000000eeadd4ba0000000015cc7f96000000e0c28229028e5c5932668d80"
+        "1127f6dcd92160c69e0702565cd972e08c803d37a69c107061c114011b86bc782260c29e39ba1addfe6d6f",
+        FOX,
+        id="xpress9",
+    ),
+    pytest.param(
+        "300010dc7a8d3d64ac36a16f0def11ff0b42434445464748494a4b4c4d4e4f505152535455565758595a411a00ffffffffffffffffff"
+        "ffffffffffffdd504b4c4d4e4f",
+        PATTERN,
+        id="xpress10",
+    ),
+    pytest.param("3800041f000100ffffffea500000000000", b"\x00" * 1024, id="lz4"),
+]
+
+
+@pytest.mark.parametrize(("cell", "plain"), CELLS)
+def test_decompress(cell: str, plain: bytes) -> None:
+    assert decompress(bytes.fromhex(cell)) == plain
+
+
+@pytest.mark.parametrize(("cell", "plain"), CELLS)
+def test_decompress_size(cell: str, plain: bytes) -> None:
+    assert decompress_size(bytes.fromhex(cell)) == len(plain)
+
+
+@pytest.mark.parametrize(("cell", "plain"), CELLS)
+def test_decompress_no_verify(cell: str, plain: bytes) -> None:
+    assert decompress(bytes.fromhex(cell), verify=False) == plain
+
+
+def test_decompress_uncompressed() -> None:
+    # Scheme 0x0 (COMPRESS_NONE) is returned unchanged.
+    buf = b"\x00raw uncompressed data"
+    assert decompress(buf) == buf
+    assert decompress_size(buf) is None
+
+
+def test_decompress_scrub() -> None:
+    # Scheme 0x4 (COMPRESS_SCRUB) is an erase marker with no recoverable data.
+    scrub = bytes([0x4 << 3]) + b"LLLL"
+    with pytest.raises(ValueError, match="SCRUB"):
+        decompress(scrub)
+    with pytest.raises(ValueError, match="SCRUB"):
+        decompress_size(scrub)
+
+
+def test_xpress9_crc_mismatch() -> None:
+    cell = bytearray.fromhex(
+        "28f83ea8df2ad7864e68010000d00200001b00060000000000eeadd4ba0000000015cc7f96000000e0c28229028e5c5932668d80"
+        "1127f6dcd92160c69e0702565cd972e08c803d37a69c107061c114011b86bc782260c29e39ba1addfe6d6f"
+    )
+    cell[1] ^= 0xFF  # corrupt the stored plaintext CRC-32C
+
+    with pytest.raises(ValueError, match="XPRESS9 plaintext CRC-32C mismatch"):
+        decompress(bytes(cell))
+
+    # verify=False skips the CRC check and still returns the plaintext.
+    assert decompress(bytes(cell), verify=False) == FOX
+
+
+def test_xpress10_crc64_mismatch() -> None:
+    cell = bytearray.fromhex(
+        "300010dc7a8d3d64ac36a16f0def11ff0b42434445464748494a4b4c4d4e4f505152535455565758595a411a00ffffffffffffffffff"
+        "ffffffffffffdd504b4c4d4e4f"
+    )
+    cell[7] ^= 0xFF  # corrupt the payload CRC-64
+
+    with pytest.raises(ValueError, match="XPRESS10 payload CRC-64 mismatch"):
+        decompress(bytes(cell))
+
+    assert decompress(bytes(cell), verify=False) == PATTERN
+
+
+def test_xpress10_crc32_mismatch() -> None:
+    cell = bytearray.fromhex(
+        "300010dc7a8d3d64ac36a16f0def11ff0b42434445464748494a4b4c4d4e4f505152535455565758595a411a00ffffffffffffffffff"
+        "ffffffffffffdd504b4c4d4e4f"
+    )
+    cell[3] ^= 0xFF  # corrupt the stored plaintext CRC-32C
+
+    with pytest.raises(ValueError, match="XPRESS10 plaintext CRC-32C mismatch"):
+        decompress(bytes(cell))
+
+    assert decompress(bytes(cell), verify=False) == PATTERN
+
+
+def test_xpress_size_mismatch() -> None:
+    # A header claiming more plaintext than the stream decodes to fails under verify.
+    cell = bytearray.fromhex("180004ffffff3f000007000ffffb03")
+    struct.pack_into("<H", cell, 1, 2048)  # header says 2048, stream yields 1024
+
+    with pytest.raises(ValueError, match="XPRESS decoded"):
+        decompress(bytes(cell))
+
+    assert len(decompress(bytes(cell), verify=False)) == 1024