Replace snappy and zstd dependencies with cramjam

.github/workflows/mypy.yml

@@ -19,8 +19,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install python-snappy
-        pip install zstd
+        pip install cramjam
         pip install mypy
     - name: Analysing the code with mypy
       run: |

.github/workflows/pylint.yml

@@ -19,8 +19,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install python-snappy
-        pip install zstd
+        pip install cramjam
         pip install pylint
     - name: Analysing the code with pylint
       run: |

.github/workflows/unittests.yml

@@ -19,8 +19,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install python-snappy
-        pip install zstd
+        pip install cramjam
     - name: Run unittests
       run: |
         python -m unittest discover -s tests -p '*.py' -t .

README.md

@@ -18,13 +18,7 @@ include:
 
 ## Installation
 
-1. [Linux] Install the snappy compression development package
-
-```
-    $ sudo apt install libsnappy-dev
-```
-
-2. Create a virtual environment and install the package
+1. Create a virtual environment and install the package
 
 ```
     $ python3 -m venv .venv
@@ -42,15 +36,9 @@ To also install the dependencies for leveldb/indexeddb plugins, run
 
 ## Installation from source
 
-1. [Linux] Install the snappy compression development package
-
-```
-    $ sudo apt install libsnappy-dev
-```
-
-2. Clone or download/unzip the repository to your local machine.
+1. Clone or download/unzip the repository to your local machine.
 
-3. Create a virtual environment and install the package
+2. Create a virtual environment and install the package
 
 ```
     $ python3 -m venv .venv
@@ -65,6 +53,27 @@ To also install the dependencies for leveldb/indexeddb plugins, run
     $ pip install '.[plugins]'
 ```
 
+## Developer Setup
+
+To set up the development environment, install the package with development dependencies:
+
+```
+    $ pip install -e '.[dev]'
+```
+
+Then install the pre-commit hooks:
+
+```
+    $ pre-commit install
+```
+
+You can manually run the pre-commit hooks on all files:
+
+```
+    $ pre-commit run --all-files
+```
+
+
 ## Usage
 
 Two CLI tools for parsing IndexedDB/LevelDB files are available after

dfindexeddb/indexeddb/chromium/blink.py

@@ -19,7 +19,7 @@
 from dataclasses import dataclass
 from typing import Any, Optional, Union
 
-import snappy
+from cramjam import snappy
 
 from dfindexeddb import errors, utils
 from dfindexeddb.indexeddb.chromium import definitions, v8
@@ -1056,5 +1056,5 @@ def FromBytes(cls, data: bytes) -> Any:
         and data[2] == definitions.COMPRESSED_WITH_SNAPPY
     ):
       # ignore the wrapped header bytes when decompressing
-      data = snappy.decompress(data[3:])
+      data = bytes(snappy.decompress_raw(data[3:]))
     return cls(data).Deserialize()

dfindexeddb/indexeddb/chromium/sqlite.py

@@ -19,8 +19,7 @@
 from typing import Any, Generator, Optional
 from dataclasses import dataclass
 
-import snappy
-import zstd
+from cramjam import snappy, zstd
 
 from dfindexeddb.indexeddb.chromium import blink
 from dfindexeddb.indexeddb.chromium import definitions
@@ -283,11 +282,11 @@ def _EnumerateCursor(
           value = blink.V8ScriptValueDecoder.FromBytes(raw_value)
         elif compression_type == definitions.DatabaseCompressionType.ZSTD:
           value = blink.V8ScriptValueDecoder.FromBytes(
-              zstd.decompress(raw_value)
+              bytes(zstd.decompress(raw_value))
           )
         elif compression_type == definitions.DatabaseCompressionType.SNAPPY:
           value = blink.V8ScriptValueDecoder.FromBytes(
-              snappy.decompress(raw_value)
+              bytes(snappy.decompress_raw(raw_value))
           )
 
       if load_blobs and raw_value is None:

dfindexeddb/indexeddb/firefox/gecko.py

@@ -21,7 +21,7 @@
 import struct
 from typing import Any, List, Optional, Tuple, Union
 
-import snappy
+from cramjam import DecompressionError, snappy
 
 from dfindexeddb import errors, utils
 from dfindexeddb.indexeddb import types
@@ -791,6 +791,7 @@ def FromBytes(cls, raw_data: bytes, base_offset: int = 0) -> Any:
     Returns:
       A python representation of the parsed JavaScript object.
     """
+    uncompressed_data: Union[bytes, bytearray]
     if raw_data.startswith(definitions.FRAME_HEADER):
       uncompressed_data = bytearray()
       pos = len(definitions.FRAME_HEADER)
@@ -806,16 +807,18 @@ def FromBytes(cls, raw_data: bytes, base_offset: int = 0) -> Any:
           uncompressed_data += raw_data[pos + 8 : pos + 8 + block_size - 4]
         else:
           try:
-            uncompressed_data += snappy.decompress(
-                raw_data[pos + 8 : pos + 8 + block_size - 4]
+            uncompressed_data += bytes(
+                snappy.decompress_raw(
+                    raw_data[pos + 8 : pos + 8 + block_size - 4]
+                )
             )
-          except snappy.UncompressError as err:
+          except DecompressionError as err:
             raise errors.ParserError("Failed to decompress", err)
         pos += block_size + 4
     else:
       try:
-        uncompressed_data = snappy.decompress(raw_data)
-      except snappy.UncompressError as err:
+        uncompressed_data = bytes(snappy.decompress_raw(raw_data))
+      except DecompressionError as err:
         raise errors.ParserError("Failed to decompress", err)
     stream = io.BytesIO(uncompressed_data)
 

dfindexeddb/leveldb/ldb.py

@@ -20,8 +20,7 @@
 from dataclasses import dataclass, field
 from typing import BinaryIO, Iterable, Tuple
 
-import snappy
-import zstd
+from cramjam import snappy, zstd
 
 from dfindexeddb.leveldb import definitions, utils
 
@@ -111,7 +110,7 @@ def IsZstdCompressed(self) -> bool:
   def GetBuffer(self) -> bytes:
     """Returns the block buffer, decompressing if required."""
     if self.IsSnappyCompressed():
-      return bytes(snappy.decompress(self.data))
+      return bytes(snappy.decompress_raw(self.data))
     if self.IsZstdCompressed():
       return bytes(zstd.decompress(self.data))
     return self.data

pyproject.toml

@@ -13,7 +13,7 @@ authors = [{ name = "Syd Pleno", email = "sydp@google.com" }]
 maintainers = [
   { name = "dfIndexeddb Developers", email = "dfindexeddb-dev@googlegroups.com" },
 ]
-dependencies = ["python-snappy>=0.6.1", "zstd>=1.5.5.1"]
+dependencies = ["cramjam"]
 readme = { file = "README.md", content-type = "text/markdown" }
 classifiers = [
   "Development Status :: 3 - Alpha",

tests/dfindexeddb/indexeddb/chromium/sqlite.py

@@ -15,6 +15,9 @@
 """Unit tests for Chromium IndexedDB encoded sqlite3 databases."""
 import datetime
 import unittest
+import pathlib
+import shutil
+import tempfile
 
 from dfindexeddb.indexeddb import types
 from dfindexeddb.indexeddb.chromium import sqlite
@@ -24,9 +27,15 @@ class ChromiumSQLiteIndexedDBTest(unittest.TestCase):
   """Unit tests for Chromium IndexedDB encoded sqlite3 databases."""
 
   def setUp(self) -> None:
-    self.reader = sqlite.DatabaseReader(
+    self._temp_dir = tempfile.TemporaryDirectory()
+    temp_dir_path = pathlib.Path(self._temp_dir.name)
+    src_db = pathlib.Path(
         "./test_data/indexeddb/chrome/osx_144_64/file__0/sample"
     )
+    self._temp_db_path = temp_dir_path / "sample"
+    shutil.copy(src_db, self._temp_db_path)
+
+    self.reader = sqlite.DatabaseReader(str(self._temp_db_path))
     expected_test_array = types.JSArray()
     for value in [123, 456, "abc", "def"]:
       expected_test_array.values.append(value)
@@ -57,6 +66,9 @@ def setUp(self) -> None:
         },
     }
 
+  def tearDown(self) -> None:
+    self._temp_dir.cleanup()
+
   def test_object_stores(self) -> None:
     """Tests the ObjectStores method."""
     object_stores = list(self.reader.ObjectStores())

tests/dfindexeddb/indexeddb/firefox/gecko.py

@@ -15,7 +15,9 @@
 """Unit tests for Gecko encoded JavaScript values."""
 import datetime
 import unittest
+from cramjam import DecompressionError
 
+from dfindexeddb import errors
 from dfindexeddb.indexeddb import types
 from dfindexeddb.indexeddb.firefox import definitions, gecko
 
@@ -495,6 +497,14 @@ def test_parse_wasm_module(self) -> None:
     parsed_value = gecko.JSStructuredCloneDecoder.FromBytes(value_bytes)
     self.assertEqual(parsed_value, expected_value)
 
+  def test_parse_corrupt_data(self) -> None:
+    """Tests that corrupt data raises ParserError due to decompression error."""
+    corrupt_bytes = b"\xff\xff\xff\xff"
+    with self.assertRaises(errors.ParserError) as context:
+      gecko.JSStructuredCloneDecoder.FromBytes(corrupt_bytes)
+    self.assertEqual(context.exception.args[0], "Failed to decompress")
+    self.assertIsInstance(context.exception.args[1], DecompressionError)
+
 
 if __name__ == "__main__":
   unittest.main()

tests/dfindexeddb/leveldb/ldb.py

@@ -15,6 +15,8 @@
 """Unit tests for LevelDB Table (.ldb) files."""
 import unittest
 
+import cramjam
+
 from dfindexeddb.leveldb import definitions, ldb
 
 
@@ -61,6 +63,38 @@ def test_range_iter(self) -> None:
     self.assertIsInstance(range_iter_records[0][1], bytes)
     self.assertEqual(range_iter_records[0][1], b"test value\x00\x00\x00\x00")
 
+  def test_zstd_block_decompression(self) -> None:
+    """Tests decompressing a ZSTD compressed block."""
+    raw_data = b"test block data"
+    compressed_data = bytes(cramjam.zstd.compress(raw_data))
+    footer = (
+        bytes([definitions.BlockCompressionType.ZSTD]) + b"\x00\x00\x00\x00"
+    )
+    block = ldb.Block(
+        offset=0,
+        block_offset=0,
+        length=len(compressed_data),
+        data=compressed_data,
+        footer=footer,
+    )
+    self.assertEqual(block.GetBuffer(), raw_data)
+
+  def test_snappy_block_decompression(self) -> None:
+    """Tests decompressing a Snappy compressed block."""
+    raw_data = b"test block data"
+    compressed_data = bytes(cramjam.snappy.compress_raw(raw_data))
+    footer = (
+        bytes([definitions.BlockCompressionType.SNAPPY]) + b"\x00\x00\x00\x00"
+    )
+    block = ldb.Block(
+        offset=0,
+        block_offset=0,
+        length=len(compressed_data),
+        data=compressed_data,
+        footer=footer,
+    )
+    self.assertEqual(block.GetBuffer(), raw_data)
+
 
 if __name__ == "__main__":
   unittest.main()