Stop storing unlocked volume keys in the case file

CHANGELOG.md

@@ -11,6 +11,15 @@ records, not how the code changed.
 
 ## [Unreleased]
 
+### Security
+
+- **Unlocked BitLocker and LUKS volume keys are no longer saved in the case
+  file.** The README always said they never were, but each unlock stored the
+  volume key, so anyone holding a case could decrypt the volume. Keys now live
+  only in memory for the session; reopening a case means unlocking again.
+  Opening a case written by an earlier version removes any stored keys and
+  wipes them from the file.
+
 ## [0.9.0] - 2026-10-02
 
 ### Added

engine/casedb.py

@@ -296,18 +296,6 @@
 
 CREATE INDEX IF NOT EXISTS idx_tm ON type_mismatches(evidence_id, part);
 
--- The key a volume was unlocked with (BitLocker's FVEK, LUKS's master key),
--- kept so reopening the case finds the volume as it was left. It is the
--- decryption key itself, so the case file holds it: anyone who can read the
--- case can read the volume. Dropped when the volume is locked on purpose or
--- the evidence is removed.
-CREATE TABLE IF NOT EXISTS volume_keys (
-    evidence_id INTEGER NOT NULL,
-    part INTEGER NOT NULL,
-    kind TEXT NOT NULL,
-    key BLOB NOT NULL,
-    saved_at TEXT NOT NULL,
-    PRIMARY KEY (evidence_id, part));
 CREATE INDEX IF NOT EXISTS idx_bm_ev ON bookmarks(evidence_id, offset);
 CREATE INDEX IF NOT EXISTS idx_tag_ev ON tagged_items(evidence_id, tag);
 CREATE INDEX IF NOT EXISTS idx_hits ON search_hits(search_id);
@@ -561,6 +549,17 @@ def __init__(self, path, name=None, examiner=None, read_only=False):
                                   timeout=30.0)
         self.db.row_factory = sqlite3.Row
         self.db.executescript(SCHEMA)
+        # Older cases kept unlocked volume keys here. A key decrypts the
+        # volume, so it is never stored in the case; drop what is left.
+        # secure_delete zeroes the freed pages and VACUUM rewrites the file,
+        # so the key bytes do not survive in free pages or the journal.
+        if self.db.execute("SELECT 1 FROM sqlite_master WHERE type='table'"
+                           " AND name='volume_keys'").fetchone():
+            self.db.execute("PRAGMA secure_delete=ON")
+            self.db.execute("DROP TABLE volume_keys")
+            self.db.commit()
+            self.db.execute("VACUUM")
+            self.db.execute("PRAGMA wal_checkpoint(TRUNCATE)")
         self._migrate_bookmarks()
         self._migrate_file_hashes()
         self._migrate_fuzzy_hash()
@@ -1079,30 +1078,7 @@ def summary(self):
 
     EVIDENCE_TABLES = ("bookmarks", "tagged_items", "saved_searches",
                        "file_hashes", "artefacts", "attack_tags",
-                       "type_mismatches", "volume_keys")
-
-    @_writes
-    def save_volume_key(self, evidence_id, part, kind, key):
-        self.db.execute(
-            "INSERT OR REPLACE INTO volume_keys VALUES (?,?,?,?,?)",
-            (evidence_id, part, kind, bytes(key), utcnow()))
-        self.db.commit()
-
-    def volume_key(self, evidence_id, part):
-        try:
-            row = self.db.execute(
-                "SELECT kind, key FROM volume_keys "
-                "WHERE evidence_id=? AND part=?", (evidence_id, part)).fetchone()
-        except sqlite3.Error:
-            return None
-        return (row["kind"], bytes(row["key"])) if row else None
-
-    @_writes
-    def forget_volume_key(self, evidence_id, part):
-        self.db.execute(
-            "DELETE FROM volume_keys WHERE evidence_id=? AND part=?",
-            (evidence_id, part))
-        self.db.commit()
+                       "type_mismatches")
 
     def evidence_holdings(self, evidence_id):
         out = {}

engine/server.py

@@ -795,35 +795,11 @@ def build():
         v = (REGISTRY.shared("vault", ev.path, offset, build)
              if REGISTRY is not None else build())
         ev.vault_cache[offset] = v
-        if v is not None and not getattr(v, "unlocked", False):
-            self._restore_unlock(offset, v, ev)
         if v is not None and getattr(v, "unlocked", False):
             ev.unlocked[offset] = v
             self._note_unlock(offset, v, ev)
         return v
 
-    def _restore_unlock(self, offset, vault, ev):
-        """A volume unlocked in an earlier session comes back unlocked: the
-        case kept the key, so nothing has to be entered again."""
-        if self.case is None or getattr(ev, "evidence_id", None) is None:
-            return
-        kept = self.case.volume_key(ev.evidence_id, offset)
-        if kept is None:
-            return
-        try:
-            if not vault.restore_key(kept[1]):
-                return
-        except Exception:
-            return
-        seen = getattr(ev, "unlock_noted", None)
-        if seen is None:
-            seen = ev.unlock_noted = set()
-        seen.add(offset)
-        try:
-            self.case.log("volume.unlock.restored", {"part": offset})
-        except Exception:
-            pass
-
     def _note_unlock(self, offset, vault, ev):
         if self.case is None:
             return
@@ -4003,14 +3979,6 @@ def run(progress):
                     if REGISTRY is not None and s.current is not None:
                         REGISTRY.drop_shared("fs", s.current.path, part)
                     s._structures = None
-                    try:
-                        key = v.export_key()
-                        if key and s.current is not None:
-                            s.case.save_volume_key(
-                                s.current.evidence_id, part,
-                                "bitlocker" if is_bde else "luks", key)
-                    except Exception:
-                        pass
                     s.case.log("volume.unlock", {
                         "part": part, "method": r.get("protector")
                         or ("key slot %s" % r.get("slot")),
@@ -4078,7 +4046,6 @@ def run(progress):
         if path == "/api/lock":
             part = int(body.get("part") or 0)
             if s.current is not None:
-                s.case.forget_volume_key(s.current.evidence_id, part)
                 if REGISTRY is not None:
                     REGISTRY.drop_shared("vault", s.current.path, part)
             s.unlocked.pop(part, None)

tests/test_bitlocker.py

@@ -213,7 +213,7 @@ def test_tweak_key_lives_in_its_own_256_bit_slot(self):
 
 
 class KeptKey(unittest.TestCase):
-    """A volume unlocked once comes back unlocked from the key the case kept."""
+    """Exported volume keys restore a vault; the case never keeps them."""
 
     @classmethod
     def setUpClass(cls):
@@ -236,15 +236,31 @@ def test_a_wrong_key_is_refused_and_leaves_the_vault_locked(self):
         self.assertFalse(v.restore_key(b"\x01" * 32))
         self.assertFalse(v.unlocked)
 
-    def test_the_case_keeps_and_forgets_the_key(self):
+    def test_the_case_never_keeps_a_volume_key(self):
         import shutil
         import tempfile
         from engine.casedb import Case
         d = tempfile.mkdtemp(prefix="strata-key-test-")
         self.addCleanup(lambda: shutil.rmtree(d, ignore_errors=True))
-        c = Case(os.path.join(d, "case"), name="t", examiner="A")
+        path = os.path.join(d, "case")
+        first = self.vault()
+        self.assertTrue(first.unlock(build.PASSWORD, kind="password")["unlocked"])
+        key = first.export_key()
+        c = Case(path, name="t", examiner="A")
+        # A case written by an older version carries the key.
+        c.db.execute("CREATE TABLE volume_keys (evidence_id INTEGER, "
+                     "part INTEGER, kind TEXT, key BLOB, saved_at TEXT)")
+        c.db.execute("INSERT INTO volume_keys VALUES (1, 2048, 'bitlocker', "
+                     "?, '')", (key,))
+        c.db.commit()
+        c.close()
+        c = Case(path, name="t", examiner="A")
         self.addCleanup(c.close)
-        c.save_volume_key(1, 2048, "bitlocker", b"k" * 32)
-        self.assertEqual(c.volume_key(1, 2048), ("bitlocker", b"k" * 32))
-        c.forget_volume_key(1, 2048)
-        self.assertIsNone(c.volume_key(1, 2048))
+        tables = [r[0] for r in c.db.execute(
+            "SELECT name FROM sqlite_master WHERE type='table'")]
+        self.assertNotIn("volume_keys", tables)
+        c.close()
+        for name in os.listdir(path):
+            if os.path.isfile(os.path.join(path, name)):
+                with open(os.path.join(path, name), "rb") as f:
+                    self.assertNotIn(key, f.read(), name)