Use improved SQLite3 to automatically open sidecar WAL files

dissect/target/loaders/itunes.py

@@ -146,7 +146,7 @@ def open(self, password: str | None = None, kek: bytes | None = None) -> None:
     def _open_manifest_db(self) -> SQLite3:
         path = self.root.joinpath("Manifest.db")
         if not self.encrypted or self.manifest["Lockdown"]["ProductVersion"] < "10.2":
-            fh = path.open("rb")
+            fh = path
         else:
             key = self.key_bag.unwrap(self.manifest["ManifestKey"])
             fh = BytesIO(aes_decrypt(path.read_bytes(), key))

dissect/target/plugins/apps/av/mcafee.py

@@ -96,8 +96,7 @@ def msc(self) -> Iterator[McAfeeMscLogRecord]:
         len_marker = len(self.MARKER_SUSPICIOUS_UDP_CONNECTION)
 
         for log_file in self.get_log_files():
-            with log_file.open() as open_log:
-                database = SQLite3(open_log)
+            with SQLite3(log_file) as database:
                 fields = defaultdict(dict)
                 fields_table = database.table(self.TABLE_FIELD)
 

dissect/target/plugins/apps/av/sophos.py

@@ -78,8 +78,7 @@ def hitmanlogs(self) -> Iterator[HitmanAlertRecord]:
         """
         if self.target.fs.path(self.LOG_SOPHOS_HITMAN).exists():
             try:
-                fh = self.target.fs.path(self.LOG_SOPHOS_HITMAN).open("rb")
-                db = SQLite3(fh)
+                db = SQLite3(self.target.fs.path(self.LOG_SOPHOS_HITMAN))
                 alerts = next(filter(lambda t: t.name == "Alerts", db.tables()))
                 for alert in alerts.rows():
                     yield HitmanAlertRecord(

dissect/target/plugins/apps/browser/chromium.py

@@ -158,7 +158,7 @@ def _iter_db(self, filename: str, subdirs: list[str] | None = None) -> Iterator[
             seen.add(db_file)
 
             try:
-                yield user, db_file, SQLite3(db_file.open())
+                yield user, db_file, SQLite3(db_file)
             except FileNotFoundError:
                 self.target.log.warning("Could not find %s file: %s", filename, db_file)
             except DBError as e:

dissect/target/plugins/apps/browser/firefox.py

@@ -170,7 +170,7 @@ def _iter_db(self, filename: str) -> Iterator[tuple[UserDetails | None, Path, SQ
                 db_file = profile_dir.joinpath(filename)
 
             try:
-                yield user_details, db_file, SQLite3(db_file.open())
+                yield user_details, db_file, SQLite3(db_file)
             except FileNotFoundError:
                 self.target.log.info("Could not find %s file: %s", filename, db_file)
             except DBError as e:
@@ -673,9 +673,7 @@ def decrypt_master_key(key4_file: Path, primary_password: bytes) -> bytes:
     # Extract neccesary information from the key4.db file. Multiple values might exist for the
     # values we are interested in. Generally the last entry will be the currently active value,
     # which is why we need to iterate every row in the table to get the last entry.
-    with key4_file.open("rb") as fh:
-        db = SQLite3(fh)
-
+    with SQLite3(key4_file) as db:
         # Get the last ``item`` (global salt) and ``item2`` (password check) values.
         if table := db.table("metadata"):
             for row in table.rows():
@@ -693,40 +691,42 @@ def decrypt_master_key(key4_file: Path, primary_password: bytes) -> bytes:
         else:
             raise ValueError(f"Missing table 'nssPrivate' in key4.db {key4_file}")
 
-    if not master_key:
-        raise ValueError(f"Password master key is not defined in key4.db {key4_file}")
+        if not master_key:
+            raise ValueError(f"Password master key is not defined in key4.db {key4_file}")
 
-    if master_key_cka != CKA_ID:
-        raise ValueError(
-            f"Password master key CKA_ID '{master_key_cka}' is not equal to expected value '{CKA_ID}' in {key4_file}"
-        )
+        if master_key_cka != CKA_ID:
+            raise ValueError(
+                f"Password master key CKA_ID '{master_key_cka}' is not equal to expected value '{CKA_ID}' in {key4_file}"  # noqa: E501
+            )
 
-    decoded_password_check: core.Sequence = core.load(password_check)
-    decoded_master_key: core.Sequence = core.load(master_key)
+        decoded_password_check: core.Sequence = core.load(password_check)
+        decoded_master_key: core.Sequence = core.load(master_key)
 
-    try:
-        decrypted_password_check, algorithm = _decrypt_master_key(decoded_password_check, primary_password, global_salt)
+        try:
+            decrypted_password_check, algorithm = _decrypt_master_key(
+                decoded_password_check, primary_password, global_salt
+            )
 
-    except EOFError:
-        raise ValueError("No primary password provided")
+        except EOFError:
+            raise ValueError("No primary password provided")
 
-    except ValueError as e:
-        raise ValueError(f"Unable to decrypt Firefox password check: {e!s}") from e
+        except ValueError as e:
+            raise ValueError(f"Unable to decrypt Firefox password check: {e!s}") from e
 
-    if not decrypted_password_check:
-        raise ValueError(f"Encountered unknown algorithm {algorithm} while decrypting Firefox master key")
+        if not decrypted_password_check:
+            raise ValueError(f"Encountered unknown algorithm {algorithm} while decrypting Firefox master key")
 
-    expected_password_check = b"password-check\x02\x02"
+        expected_password_check = b"password-check\x02\x02"
 
-    if decrypted_password_check != expected_password_check:
-        log.debug("Expected %s but got %s", expected_password_check, decrypted_password_check)
-        raise ValueError("Master key decryption failed. Provided password could be missing or incorrect")
+        if decrypted_password_check != expected_password_check:
+            log.debug("Expected %s but got %s", expected_password_check, decrypted_password_check)
+            raise ValueError("Master key decryption failed. Provided password could be missing or incorrect")
 
-    decrypted, algorithm = _decrypt_master_key(decoded_master_key, primary_password, global_salt)
+        decrypted, algorithm = _decrypt_master_key(decoded_master_key, primary_password, global_salt)
 
-    block_size = 16 if algos.EncryptionAlgorithmId.map(algorithm) == "pbes2" else 8
+        block_size = 16 if algos.EncryptionAlgorithmId.map(algorithm) == "pbes2" else 8
 
-    return unpad(decrypted, block_size)
+        return unpad(decrypted, block_size)
 
 
 def decrypt_value(b64_ciphertext: str, key: bytes) -> bytes | None:

dissect/target/plugins/apps/container/podman.py

@@ -175,7 +175,7 @@ def _find_containers_sqlite(self, path: Path) -> Iterator[PodmanContainerRecord]
         """
 
         try:
-            db = SQLite3(path.open("rb"))
+            db = SQLite3(path)
         except (ValueError, DBError) as e:
             self.target.log.warning("Unable to read Podman database %s: %s", path, e)
             self.target.log.debug("", exc_info=e)

dissect/target/plugins/os/unix/esxi/configstore.py

@@ -1,14 +1,16 @@
 from __future__ import annotations
 
 import json as jsonlib
-from typing import TYPE_CHECKING, Any, BinaryIO
+from typing import TYPE_CHECKING, Any
 
 from dissect.database.sqlite3 import SQLite3
 
 from dissect.target.exceptions import UnsupportedPluginError
 from dissect.target.plugin import Plugin, internal
 
 if TYPE_CHECKING:
+    from pathlib import Path
+
     from dissect.target.target import Target
 
 
@@ -25,8 +27,7 @@ def __init__(self, target: Target):
         # It's made available at /etc/vmware/configstore/current-store-1 during boot, but stored at
         # the path used below in local.tgz
         if (path := target.fs.path("/var/lib/vmware/configstore/backup/current-store-1")).exists():
-            with path.open("rb") as fh:
-                self._configstore = parse_config_store(fh)
+            self._configstore = parse_config_store(path)
 
     def check_compatible(self) -> None:
         if self.target.os != "esxi":
@@ -41,43 +42,42 @@ def get(self, key: str, default: Any = None) -> dict[str, Any]:
         return self._configstore.get(key, default)
 
 
-def parse_config_store(fh: BinaryIO) -> dict[str, Any]:
-    db = SQLite3(fh)
-
-    store = {}
-
-    if table := db.table("Config"):
-        for row in table.rows():
-            component_name = row.Component
-            config_group_name = row.ConfigGroup
-            value_group_name = row.Name
-            identifier_name = row.Identifier
-
-            if component_name not in store:
-                store[component_name] = {}
-            component = store[component_name]
-
-            if config_group_name not in component:
-                component[config_group_name] = {}
-            config_group = component[config_group_name]
-
-            if value_group_name not in config_group:
-                config_group[value_group_name] = {}
-            value_group = config_group[value_group_name]
-
-            if identifier_name not in value_group:
-                value_group[identifier_name] = {}
-            identifier = value_group[identifier_name]
-
-            identifier["modified_time"] = row.ModifiedTime
-            identifier["creation_time"] = row.CreationTime
-            identifier["version"] = row.Version
-            identifier["success"] = row.Success
-            identifier["auto_conf_value"] = jsonlib.loads(row.AutoConfValue) if row.AutoConfValue else None
-            identifier["user_value"] = jsonlib.loads(row.UserValue) if row.UserValue else None
-            identifier["vital_value"] = jsonlib.loads(row.VitalValue) if row.VitalValue else None
-            identifier["cached_value"] = jsonlib.loads(row.CachedValue) if row.CachedValue else None
-            identifier["desired_value"] = jsonlib.loads(row.DesiredValue) if row.DesiredValue else None
-            identifier["revision"] = row.Revision
-
-    return store
+def parse_config_store(path: Path) -> dict[str, Any]:
+    with SQLite3(path) as db:
+        store = {}
+
+        if table := db.table("Config"):
+            for row in table.rows():
+                component_name = row.Component
+                config_group_name = row.ConfigGroup
+                value_group_name = row.Name
+                identifier_name = row.Identifier
+
+                if component_name not in store:
+                    store[component_name] = {}
+                component = store[component_name]
+
+                if config_group_name not in component:
+                    component[config_group_name] = {}
+                config_group = component[config_group_name]
+
+                if value_group_name not in config_group:
+                    config_group[value_group_name] = {}
+                value_group = config_group[value_group_name]
+
+                if identifier_name not in value_group:
+                    value_group[identifier_name] = {}
+                identifier = value_group[identifier_name]
+
+                identifier["modified_time"] = row.ModifiedTime
+                identifier["creation_time"] = row.CreationTime
+                identifier["version"] = row.Version
+                identifier["success"] = row.Success
+                identifier["auto_conf_value"] = jsonlib.loads(row.AutoConfValue) if row.AutoConfValue else None
+                identifier["user_value"] = jsonlib.loads(row.UserValue) if row.UserValue else None
+                identifier["vital_value"] = jsonlib.loads(row.VitalValue) if row.VitalValue else None
+                identifier["cached_value"] = jsonlib.loads(row.CachedValue) if row.CachedValue else None
+                identifier["desired_value"] = jsonlib.loads(row.DesiredValue) if row.DesiredValue else None
+                identifier["revision"] = row.Revision
+
+        return store

dissect/target/plugins/os/unix/linux/debian/proxmox/_os.py

@@ -18,6 +18,8 @@
 from dissect.target.plugins.os.unix.linux.debian._os import DebianPlugin
 
 if TYPE_CHECKING:
+    from pathlib import Path
+
     from typing_extensions import Self
 
     from dissect.target.target import Target
@@ -37,8 +39,7 @@ def create(cls, target: Target, sysvol: Filesystem) -> Self:
         obj = super().create(target, sysvol)
 
         if (config_db := target.fs.path("/var/lib/pve-cluster/config.db")).exists():
-            with config_db.open("rb") as fh:
-                vfs = _create_pmxcfs(fh, obj.hostname)
+            vfs = _create_pmxcfs(config_db, obj.hostname)
 
             target.fs.mount("/etc/pve", vfs)
 
@@ -63,41 +64,40 @@ def os(self) -> str:
 DT_REG = 8
 
 
-def _create_pmxcfs(fh: BinaryIO, hostname: str | None = None) -> VirtualFilesystem:
+def _create_pmxcfs(fh: Path, hostname: str | None = None) -> VirtualFilesystem:
     # https://pve.proxmox.com/wiki/Proxmox_Cluster_File_System_(pmxcfs)
-    db = SQLite3(fh)
-
-    entries = {row.inode: row for row in db.table("tree")}
-
-    vfs = VirtualFilesystem()
-    for entry in entries.values():
-        if entry.type == DT_DIR:
-            cls = ProxmoxConfigDirectoryEntry
-        elif entry.type == DT_REG:
-            cls = ProxmoxConfigFileEntry
-        else:
-            raise ValueError(f"Unknown pmxcfs file type: {entry.type}")
-
-        parts = []
-        current = entry
-        while current.parent != 0:
+    with SQLite3(fh) as db:
+        entries = {row.inode: row for row in db.table("tree")}
+
+        vfs = VirtualFilesystem()
+        for entry in entries.values():
+            if entry.type == DT_DIR:
+                cls = ProxmoxConfigDirectoryEntry
+            elif entry.type == DT_REG:
+                cls = ProxmoxConfigFileEntry
+            else:
+                raise ValueError(f"Unknown pmxcfs file type: {entry.type}")
+
+            parts = []
+            current = entry
+            while current.parent != 0:
+                parts.append(current.name)
+                current = entries[current.parent]
             parts.append(current.name)
-            current = entries[current.parent]
-        parts.append(current.name)
 
-        path = "/".join(parts[::-1])
-        vfs.map_file_entry(path, cls(vfs, path, entry))
+            path = "/".join(parts[::-1])
+            vfs.map_file_entry(path, cls(vfs, path, entry))
 
-    if hostname:
-        node_root = vfs.path(f"nodes/{hostname}")
-        vfs.symlink(str(node_root), "local")
-        vfs.symlink(str(node_root / "lxc"), "lxc")
-        vfs.symlink(str(node_root / "openvz"), "openvz")
-        vfs.symlink(str(node_root / "qemu-server"), "qemu-server")
+        if hostname:
+            node_root = vfs.path(f"nodes/{hostname}")
+            vfs.symlink(str(node_root), "local")
+            vfs.symlink(str(node_root / "lxc"), "lxc")
+            vfs.symlink(str(node_root / "openvz"), "openvz")
+            vfs.symlink(str(node_root / "qemu-server"), "qemu-server")
 
-    # TODO: .version, .members, .vmlist, maybe .clusterlog and .rrd?
+        # TODO: .version, .members, .vmlist, maybe .clusterlog and .rrd?
 
-    return vfs
+        return vfs
 
 
 class ProxmoxConfigFileEntry(VirtualFile):