From d18376828faddc6c2129b0d90a477786ec8db629 Mon Sep 17 00:00:00 2001 From: sck_0 Date: Tue, 23 Jun 2026 10:44:49 +0200 Subject: [PATCH 1/2] Avoid recovered output basename collisions --- README.md | 6 +- lockbit-rescue.py | 123 ++++++++++++++++++++++++++++++++++--- tests/test_output_paths.py | 65 ++++++++++++++++++++ 3 files changed, 183 insertions(+), 11 deletions(-) create mode 100644 tests/test_output_paths.py diff --git a/README.md b/README.md index eff7d8c..acb5532 100644 --- a/README.md +++ b/README.md @@ -110,14 +110,16 @@ python3 lockbit-rescue.py SOURCE_DIR OUTPUT_DIR OUTPUT_DIR/ ├── group_a1b2c3d4e5f6/ # one folder per encryption batch │ ├── photo1.jpg -│ ├── docs/report.pdf # original sub-paths flattened — see note +│ ├── report.pdf # original sub-paths flattened — see note +│ ├── report__9f2a1c44.pdf # duplicate basename disambiguated by source path │ └── ... ├── group_f0e9d8c7b6a5/ │ └── ... +├── manifest.csv # source -> output mapping and per-file status └── .scratch/ # temporary working files (safe to delete after) ``` -> **Note**: filenames inside `group_*/` keep their original *basename*, not their original full path. If you need to map a recovered file back to the original directory tree, cross-reference by basename with your encrypted source. A future version may emit a `manifest.csv`. +> **Note**: filenames inside `group_*/` keep their original *basename*, not their original full path. When multiple source files in the same group share a basename, the tool appends a stable hash suffix so one recovered file cannot hide another. Use `manifest.csv` to map every source path to its recovered output and status. ### Verifying results diff --git a/lockbit-rescue.py b/lockbit-rescue.py index 56d0012..c3a3b62 100644 --- a/lockbit-rescue.py +++ b/lockbit-rescue.py @@ -17,7 +17,9 @@ 5) Decrypt all other files in the group whose footer-encryption-info length fits within the recovered keystream 6) Save recovered files to OUTPUT_DIR/group_/ - 7) Verify each output with libmagic and skip writes for raw "data" results + (suffixing duplicate basenames so outputs do not collide) + 7) Write manifest.csv with source/output/status mapping + 8) Verify each output with libmagic and skip writes for raw "data" results Requires: - The `stream-reuse` binary (built from yohanes/lockbit-v3-linux-decryptor) @@ -39,6 +41,7 @@ import argparse import collections +import csv import hashlib import os import shutil @@ -93,6 +96,10 @@ # Coverage formula derived empirically: see TECHNICAL.md. COVERAGE_OFFSET = 18 COVERAGE_BASE_FROM_FEI = 82 # bytes consumed by fixed metadata +MANIFEST_COLUMNS = [ + "kek", "source_path", "output_path", "original_name", + "fei_len", "size", "status", "magic", +] def detect_extension(source: Path, sample_limit: int = 5000) -> str: @@ -156,6 +163,18 @@ def copy_with_progress(src: Path, dst: Path, label: str, position: int = 2): bar.close() +def copy_atomic_with_progress(src: Path, dst: Path, label: str, position: int = 2): + tmp = dst.with_name(f".{dst.name}.tmp") + try: + copy_with_progress(src, tmp, label, position) + os.replace(tmp, dst) + finally: + try: + tmp.unlink() + except FileNotFoundError: + pass + + def read_footer(path: Path): """Return (fei_len, kek_blob) from the last 134 bytes of an encrypted file.""" with open(path, "rb") as f: @@ -170,6 +189,51 @@ def kek_fingerprint(kek_blob: bytes) -> str: return hashlib.md5(kek_blob).hexdigest()[:12] +def source_relpath(source: Path, path: str) -> str: + p = Path(path).resolve() + try: + return str(p.relative_to(source)) + except ValueError: + return p.name + + +def collision_safe_name(original_name: str, rel_path: str, collides: bool) -> str: + if not collides: + return original_name + suffix = hashlib.sha1(rel_path.encode("utf-8", "surrogateescape")).hexdigest()[:8] + p = Path(original_name) + if p.suffix: + return f"{p.stem}__{suffix}{p.suffix}" + return f"{original_name}__{suffix}" + + +def build_output_paths(plans, source: Path, output: Path, ransom_ext: str): + """Return {(kek, encrypted_path): output_path}, suffixing duplicate basenames.""" + output_paths = {} + collisions = 0 + for _, kek, _, targets in plans: + name_keys = {tfname: tfname[: -len(ransom_ext)].casefold() for _, tfname, _, _ in targets} + base_counts = collections.Counter(name_keys[tfname] for _, tfname, _, _ in targets) + collisions += sum(c for c in base_counts.values() if c > 1) + group_out = output / f"group_{kek}" + for _, tfname, tpath, _ in targets: + original = tfname[: -len(ransom_ext)] + rel_path = source_relpath(source, tpath) + out_name = collision_safe_name(original, rel_path, base_counts[name_keys[tfname]] > 1) + output_paths[(kek, tpath)] = group_out / out_name + return output_paths, collisions + + +def append_manifest(manifest: Path, row: dict): + manifest.parent.mkdir(parents=True, exist_ok=True) + needs_header = not manifest.exists() or manifest.stat().st_size == 0 + with open(manifest, "a", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=MANIFEST_COLUMNS) + if needs_header: + writer.writeheader() + writer.writerow({k: row.get(k, "") for k in MANIFEST_COLUMNS}) + + def scan(source: Path, ransom_ext: str, common_exts, no_extension_filter: bool, min_size: int, max_size: int): """Walk source, group encrypted files by KEK fingerprint.""" @@ -333,10 +397,15 @@ def main(): plans = build_plan(groups, args.ext) total_targets = sum(p[0] for p in plans) no_oracle = len(groups) - len(plans) + output_paths, collision_targets = build_output_paths(plans, source, output, args.ext) + manifest = output / "manifest.csv" print(f"[+] Plan: {len(plans)} decryptable groups / {len(groups)} total") print(f" ({no_oracle} groups skipped — no oracle file with long enough filename)") print(f" targets to attempt: {total_targets}") + if collision_targets: + print(f" basename collisions protected: {collision_targets} target(s)") print(f" output: {output}") + print(f" manifest: {manifest}") if total_targets == 0: print("[!] Nothing to decrypt. Probably no group has a long-named oracle.") return @@ -344,11 +413,9 @@ def main(): # --- Resume bookkeeping --- already = 0 for _, kek, _, targets in plans: - gdir = output / f"group_{kek}" - if gdir.is_dir(): - for _, tfname, _, _ in targets: - if (gdir / tfname[: -len(args.ext)]).exists(): - already += 1 + for _, _, tpath, _ in targets: + if output_paths[(kek, tpath)].exists(): + already += 1 if already: print(f"[i] Resume: {already} files already in output, will skip") @@ -364,7 +431,7 @@ def main(): group_out.mkdir(parents=True, exist_ok=True) # Skip group if fully done - existing = sum(1 for _, tf, _, _ in targets if (group_out / tf[:-len(args.ext)]).exists()) + existing = sum(1 for _, _, tp, _ in targets if output_paths[(kek, tp)].exists()) if existing == len(targets): print(f"[GROUP {gi+1}/{len(plans)}] {kek} already complete, skip") continue @@ -378,6 +445,14 @@ def main(): copy_with_progress(Path(oracle_path), local_oracle, f" copy oracle {fmt_size(oracle_sz)}") except Exception as e: print(f" [!] oracle copy failed: {e}") + for fei_len, tfname, tpath, tsz in targets: + torig = tfname[: -len(args.ext)] + append_manifest(manifest, { + "kek": kek, "source_path": tpath, + "output_path": output_paths[(kek, tpath)], + "original_name": torig, "fei_len": fei_len, "size": tsz, + "status": "oracle_copy_failed", + }) overall.update(len(targets)) continue @@ -387,13 +462,18 @@ def main(): t_start = time.time() for (fei_len, tfname, tpath, tsz) in grp_bar: torig = tfname[: -len(args.ext)] - out_path = group_out / torig + out_path = output_paths[(kek, tpath)] short = (torig[:35] + "...") if len(torig) > 35 else torig grp_bar.set_postfix({"cur": short, "sz": fmt_size(tsz), "ok": grp_ok, "fail": grp_fail}) if out_path.exists(): grp_ok += 1 + append_manifest(manifest, { + "kek": kek, "source_path": tpath, "output_path": out_path, + "original_name": torig, "fei_len": fei_len, "size": tsz, + "status": "skipped_existing", + }) overall.update(1) continue @@ -402,6 +482,11 @@ def main(): copy_with_progress(Path(tpath), local_target, f" fetch {short}") except Exception: grp_fail += 1 + append_manifest(manifest, { + "kek": kek, "source_path": tpath, "output_path": out_path, + "original_name": torig, "fei_len": fei_len, "size": tsz, + "status": "copy_failed", + }) overall.update(1) continue @@ -409,19 +494,39 @@ def main(): oracle_orig, scratch, args.timeout) if decrypted is None: grp_fail += 1 + append_manifest(manifest, { + "kek": kek, "source_path": tpath, "output_path": out_path, + "original_name": torig, "fei_len": fei_len, "size": tsz, + "status": "decrypt_failed", + }) else: ftype = libmagic(decrypted) if is_bad_decrypt(ftype): grp_fail += 1 + append_manifest(manifest, { + "kek": kek, "source_path": tpath, "output_path": out_path, + "original_name": torig, "fei_len": fei_len, "size": tsz, + "status": "suspect", "magic": ftype, + }) try: decrypted.unlink() except: pass else: try: - copy_with_progress(decrypted, out_path, f" save {short}") + copy_atomic_with_progress(decrypted, out_path, f" save {short}") decrypted.unlink() grp_ok += 1 + append_manifest(manifest, { + "kek": kek, "source_path": tpath, "output_path": out_path, + "original_name": torig, "fei_len": fei_len, "size": tsz, + "status": "recovered", "magic": ftype, + }) except Exception: grp_fail += 1 + append_manifest(manifest, { + "kek": kek, "source_path": tpath, "output_path": out_path, + "original_name": torig, "fei_len": fei_len, "size": tsz, + "status": "save_failed", "magic": ftype, + }) try: decrypted.unlink() except: pass diff --git a/tests/test_output_paths.py b/tests/test_output_paths.py new file mode 100644 index 0000000..eefb234 --- /dev/null +++ b/tests/test_output_paths.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python3 +import csv +import importlib.util +import sys +import tempfile +import types +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +fake_tqdm = types.ModuleType("tqdm") +fake_tqdm.tqdm = lambda iterable=None, *args, **kwargs: iterable if iterable is not None else [] +sys.modules.setdefault("tqdm", fake_tqdm) +spec = importlib.util.spec_from_file_location("lockbit_rescue", ROOT / "lockbit-rescue.py") +lockbit_rescue = importlib.util.module_from_spec(spec) +spec.loader.exec_module(lockbit_rescue) + + +def test_duplicate_basenames_get_stable_unique_output_paths(): + source = Path("/source").resolve() + output = Path("/recovered") + plans = [ + ( + 3, + "abc123", + None, + [ + (90, "report.pdf.lockedext", "/source/a/report.pdf.lockedext", 100), + (91, "report.pdf.lockedext", "/source/b/report.pdf.lockedext", 200), + (92, "photo.jpg.lockedext", "/source/a/photo.jpg.lockedext", 300), + ], + ) + ] + + first, collisions = lockbit_rescue.build_output_paths(plans, source, output, ".lockedext") + second, _ = lockbit_rescue.build_output_paths(plans, source, output, ".lockedext") + + a = first[("abc123", "/source/a/report.pdf.lockedext")] + b = first[("abc123", "/source/b/report.pdf.lockedext")] + photo = first[("abc123", "/source/a/photo.jpg.lockedext")] + + assert collisions == 2 + assert a != b + assert a.name.startswith("report__") and a.suffix == ".pdf" + assert b.name.startswith("report__") and b.suffix == ".pdf" + assert photo == output / "group_abc123" / "photo.jpg" + assert first == second + + +def test_manifest_has_one_header_and_expected_rows(): + with tempfile.TemporaryDirectory() as tmp: + manifest = Path(tmp) / "manifest.csv" + lockbit_rescue.append_manifest(manifest, {"kek": "abc", "status": "recovered"}) + lockbit_rescue.append_manifest(manifest, {"kek": "def", "status": "suspect"}) + + with open(manifest, newline="", encoding="utf-8") as f: + rows = list(csv.DictReader(f)) + + assert [row["kek"] for row in rows] == ["abc", "def"] + assert [row["status"] for row in rows] == ["recovered", "suspect"] + + +if __name__ == "__main__": + test_duplicate_basenames_get_stable_unique_output_paths() + test_manifest_has_one_header_and_expected_rows() From 2a0bf936447098299bdb73f26da54d102a3b3d01 Mon Sep 17 00:00:00 2001 From: sck_0 Date: Tue, 23 Jun 2026 10:55:33 +0200 Subject: [PATCH 2/2] Address Copilot review feedback --- lockbit-rescue.py | 19 ++++++++++++------- tests/test_output_paths.py | 7 ++++++- 2 files changed, 18 insertions(+), 8 deletions(-) diff --git a/lockbit-rescue.py b/lockbit-rescue.py index c3a3b62..62e1ebd 100644 --- a/lockbit-rescue.py +++ b/lockbit-rescue.py @@ -164,7 +164,8 @@ def copy_with_progress(src: Path, dst: Path, label: str, position: int = 2): def copy_atomic_with_progress(src: Path, dst: Path, label: str, position: int = 2): - tmp = dst.with_name(f".{dst.name}.tmp") + tmp_id = hashlib.sha1(str(dst).encode("utf-8", "surrogateescape")).hexdigest()[:12] + tmp = dst.with_name(f".tmp-{tmp_id}") try: copy_with_progress(src, tmp, label, position) os.replace(tmp, dst) @@ -190,11 +191,11 @@ def kek_fingerprint(kek_blob: bytes) -> str: def source_relpath(source: Path, path: str) -> str: - p = Path(path).resolve() + p = Path(path) try: return str(p.relative_to(source)) except ValueError: - return p.name + return str(p) def collision_safe_name(original_name: str, rel_path: str, collides: bool) -> str: @@ -445,15 +446,20 @@ def main(): copy_with_progress(Path(oracle_path), local_oracle, f" copy oracle {fmt_size(oracle_sz)}") except Exception as e: print(f" [!] oracle copy failed: {e}") + pending = 0 for fei_len, tfname, tpath, tsz in targets: torig = tfname[: -len(args.ext)] + out_path = output_paths[(kek, tpath)] + status = "skipped_existing" if out_path.exists() else "oracle_copy_failed" + if status != "skipped_existing": + pending += 1 append_manifest(manifest, { "kek": kek, "source_path": tpath, - "output_path": output_paths[(kek, tpath)], + "output_path": out_path, "original_name": torig, "fei_len": fei_len, "size": tsz, - "status": "oracle_copy_failed", + "status": status, }) - overall.update(len(targets)) + overall.update(pending) continue grp_ok = grp_fail = 0 @@ -474,7 +480,6 @@ def main(): "original_name": torig, "fei_len": fei_len, "size": tsz, "status": "skipped_existing", }) - overall.update(1) continue local_target = scratch / f"_target{args.ext}" diff --git a/tests/test_output_paths.py b/tests/test_output_paths.py index eefb234..fe8f29d 100644 --- a/tests/test_output_paths.py +++ b/tests/test_output_paths.py @@ -10,7 +10,7 @@ ROOT = Path(__file__).resolve().parents[1] fake_tqdm = types.ModuleType("tqdm") fake_tqdm.tqdm = lambda iterable=None, *args, **kwargs: iterable if iterable is not None else [] -sys.modules.setdefault("tqdm", fake_tqdm) +sys.modules["tqdm"] = fake_tqdm spec = importlib.util.spec_from_file_location("lockbit_rescue", ROOT / "lockbit-rescue.py") lockbit_rescue = importlib.util.module_from_spec(spec) spec.loader.exec_module(lockbit_rescue) @@ -60,6 +60,11 @@ def test_manifest_has_one_header_and_expected_rows(): assert [row["status"] for row in rows] == ["recovered", "suspect"] +def test_source_relpath_falls_back_to_full_path_for_outside_paths(): + assert lockbit_rescue.source_relpath(Path("/source"), "/outside/report.pdf") == "/outside/report.pdf" + + if __name__ == "__main__": test_duplicate_basenames_get_stable_unique_output_paths() test_manifest_has_one_header_and_expected_rows() + test_source_relpath_falls_back_to_full_path_for_outside_paths()