reveal: keep the original stego file when nothing was extracted

.github/workflows/ci.yml

@@ -40,3 +40,4 @@ jobs:
           .\scripts\test-registry-delete-safety.ps1
           python .\scripts\test-cli-temp-archive.py
           python .\scripts\test-mp4-free-atom-format.py
+          python .\scripts\test-reveal-keeps-original.py

Steganographier.py

@@ -2415,6 +2415,37 @@ def load_passwords(self):
 
         return passwords
 
+    # 7-Zip 分两步解压时创建的临时目录前缀（见 _extract_with_7zip）。那些目录里的是中间
+    # 产物而非解出来的结果，确认“有没有解出东西”时必须排除；万一清理失败，也不能把中间
+    # 产物当成结果，否则又会在什么都没解出来时删掉原件。
+    EXTRACT_TEMP_DIR_PREFIXES = ("7z_", "7z_unzip_")
+
+    def snapshot_output_files(self, output_dir, exclude_path=None):
+        """输出目录内的文件快照：相对路径 -> (大小, mtime 纳秒)。
+
+        用于判断一次解除隐写到底有没有写出文件。不能用某个解压方法的返回值代替：
+        7-Zip 是由子进程直接把文件写进输出目录的，进程内的计数覆盖不到它；而空压缩包、
+        只含目录项的压缩包、成员名全部被过滤掉的压缩包，都会让方法返回成功却什么都没写。
+        """
+        snapshot = {}
+        excluded = os.path.normcase(os.path.abspath(exclude_path)) if exclude_path else None
+        for root, _dirs, files in os.walk(output_dir):
+            relative_root = os.path.relpath(root, output_dir)
+            parts = [] if relative_root == "." else relative_root.split(os.sep)
+            if any(part.startswith(self.EXTRACT_TEMP_DIR_PREFIXES) for part in parts):
+                continue
+            for file_name in files:
+                full_path = os.path.join(root, file_name)
+                if excluded and os.path.normcase(os.path.abspath(full_path)) == excluded:
+                    continue
+                try:
+                    status = os.stat(full_path)
+                except OSError:
+                    continue
+                key = file_name if relative_root == "." else os.path.join(relative_root, file_name)
+                snapshot[key] = (status.st_size, status.st_mtime_ns)
+        return snapshot
+
     def reveal_file(self, input_file_path, password=None, type_option_var=None, delete_original=True, auto_rename_on_conflict=None):
         """
         智能解除隐写函数 - 优先使用7-Zip高速解压
@@ -2436,6 +2467,8 @@ def reveal_file(self, input_file_path, password=None, type_option_var=None, dele
 
         output_dir = os.path.dirname(input_file_path)
         if not output_dir: output_dir = '.' # 如果是空字符串,使用当前目录
+        # 解除前后各拍一次快照，据此判断这次到底有没有写出文件（见 snapshot_output_files）。
+        files_before_reveal = self.snapshot_output_files(output_dir, input_file_path)
 
         success = False
         successful_method = None
@@ -2545,7 +2578,13 @@ def reveal_file(self, input_file_path, password=None, type_option_var=None, dele
                 continue
 
         # 处理结果
-        if success:
+        if success and self.snapshot_output_files(output_dir, input_file_path) == files_before_reveal:
+            # 有方法报了成功，但输出目录里没多出任何文件：压缩包里只有目录项、成员名全部
+            # 被 sanitize_path 过滤掉、空压缩包，或 7-Zip 解到临时目录后搬运失败。
+            # 此时删原件就是纯粹的数据丢失——内容只存在于这个隐写文件里——所以保留它。
+            self.log(f"解压方法报告成功（{successful_method}），但输出目录里没有出现任何新文件；"
+                     "已保留原始隐写文件以免数据丢失")
+        elif success:
             if delete_original:
                 try:
                     os.remove(input_file_path)
@@ -2819,7 +2858,7 @@ def _extract_with_zipfile(self, input_file_path, password_bytes, total_size_hidd
                                     if self.progress_callback:
                                         self.progress_callback(processed_size, total_size_hidden)
                                     last_progress_update = processed_size
-                                    
+
                     except (OSError, IOError) as e:
                         self.log(f"无法创建文件 {output_file_path}: {e}")
                         continue
@@ -2911,7 +2950,7 @@ def _extract_with_pyzipper(self, input_file_path, password_bytes, total_size_hid
                                     if self.progress_callback:
                                         self.progress_callback(processed_size, total_size_hidden)
                                     last_progress_update = processed_size
-                                    
+
                     except (OSError, IOError) as e:
                         self.log(f"无法创建文件 {output_file_path}: {e}")
                         continue

scripts/test-reveal-keeps-original.py

@@ -0,0 +1,167 @@
+"""Regression tests for deleting the original stego file after a reveal.
+
+Guards a data-loss path: reveal_file() deleted the input whenever an extraction
+method reported success, without checking that anything had actually reached the
+disk. An archive whose members are all directory entries, an empty archive, or
+names that sanitize_path() rejects all end with a method returning True and zero
+files written -- and the original was removed anyway. The archive inside the
+stego file is the only remaining copy once that happens.
+
+The two halves are both asserted, because a guard that simply stopped deleting
+anything would pass the first half and silently break the documented default
+(delete_original_after_reveal defaults to True).
+"""
+
+import os
+import shutil
+import struct
+import subprocess
+import sys
+import tempfile
+import zipfile
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+MODULE_PATH = os.path.join(REPO_ROOT, "Steganographier.py")
+
+PAYLOAD_NAME = "payload.bin"
+PAYLOAD_SIZE = 28 * 1024
+
+
+def build_payload(seed=20260921):
+    import random
+
+    rng = random.Random(seed)
+    return bytes(rng.getrandbits(8) for _ in range(PAYLOAD_SIZE))
+
+
+def write_synthetic_cover(path):
+    """A minimal but well-formed top-level atom chain."""
+
+    def atom(atom_type, payload=b""):
+        return struct.pack(">I4s", 8 + len(payload), atom_type) + payload
+
+    data = (
+        atom(b"ftyp", b"isom" + struct.pack(">I", 512) + b"isomiso2mp41")
+        + atom(b"free")
+        + atom(b"mdat", b"\x00" * 2048)
+        + atom(b"moov", b"\x00" * 64)
+    )
+    with open(path, "wb") as handle:
+        handle.write(data)
+
+
+def append_directory_only_archive(cover_path, stego_path):
+    """Append an archive whose only member is a directory entry.
+
+    This is a legitimate thing for a user to have produced (zipping an empty
+    folder), and it is the shape that makes an extractor report success while
+    writing no files.
+    """
+    with open(stego_path, "wb") as handle:
+        with open(cover_path, "rb") as cover:
+            shutil.copyfileobj(cover, handle)
+        with zipfile.ZipFile(handle, "w", zipfile.ZIP_DEFLATED) as archive:
+            archive.writestr("empty-folder/", b"")
+
+
+def run_cli(*arguments):
+    return subprocess.run(
+        [sys.executable, MODULE_PATH] + list(arguments),
+        cwd=REPO_ROOT,
+        capture_output=True,
+        timeout=300,
+    )
+
+
+def hide(payload_path, cover_path, output_path):
+    result = run_cli(
+        "-i", payload_path, "-c", cover_path, "-t", "mp4", "-o", output_path, "--no-log"
+    )
+    if result.returncode != 0:
+        raise AssertionError(
+            "hide failed (%d)\n%s" % (result.returncode, result.stderr.decode("utf-8", "replace"))
+        )
+
+
+def test_nothing_written_keeps_the_original(work_root):
+    cover_path = os.path.join(work_root, "cover-empty.zip-case.mp4")
+    write_synthetic_cover(cover_path)
+    stego_path = os.path.join(work_root, "dir-only.mp4")
+    append_directory_only_archive(cover_path, stego_path)
+
+    result = run_cli("-r", "-i", stego_path, "--no-log")
+    if result.returncode != 0:
+        raise AssertionError(
+            "reveal failed (%d)\n%s" % (result.returncode, result.stderr.decode("utf-8", "replace"))
+        )
+    if not os.path.exists(stego_path):
+        raise AssertionError(
+            "the stego file was deleted even though no file was extracted from it; "
+            "the archive inside it was the only remaining copy"
+        )
+    print("  ok   nothing extracted -> original stego file kept")
+
+
+def test_normal_reveal_still_deletes_the_original(work_root):
+    payload_path = os.path.join(work_root, PAYLOAD_NAME)
+    with open(payload_path, "wb") as handle:
+        handle.write(build_payload())
+    cover_path = os.path.join(work_root, "cover-normal.mp4")
+    write_synthetic_cover(cover_path)
+
+    # Its own directory: reveal writes the payload next to its input.
+    case_dir = os.path.join(work_root, "normal")
+    os.makedirs(case_dir)
+    stego_path = os.path.join(case_dir, "stego.mp4")
+    hide(payload_path, cover_path, stego_path)
+
+    result = run_cli("-r", "-i", stego_path, "--no-log")
+    if result.returncode != 0:
+        raise AssertionError(
+            "reveal failed (%d)\n%s" % (result.returncode, result.stderr.decode("utf-8", "replace"))
+        )
+    extracted = os.path.join(case_dir, PAYLOAD_NAME)
+    if not os.path.exists(extracted):
+        raise AssertionError("reveal produced no %s" % PAYLOAD_NAME)
+    if open(extracted, "rb").read() != build_payload():
+        raise AssertionError("revealed file differs from the original payload")
+    if os.path.exists(stego_path):
+        raise AssertionError(
+            "the stego file survived a successful reveal; the default is to delete it "
+            "and this guard must not change that"
+        )
+    print("  ok   successful reveal still deletes the original")
+
+
+def test_keep_original_flag_still_works(work_root):
+    payload_path = os.path.join(work_root, PAYLOAD_NAME)
+    cover_path = os.path.join(work_root, "cover-normal.mp4")
+    case_dir = os.path.join(work_root, "keep")
+    os.makedirs(case_dir)
+    stego_path = os.path.join(case_dir, "stego.mp4")
+    hide(payload_path, cover_path, stego_path)
+
+    result = run_cli("-r", "-i", stego_path, "--keep-original", "--no-log")
+    if result.returncode != 0:
+        raise AssertionError("reveal --keep-original failed (%d)" % result.returncode)
+    if not os.path.exists(stego_path):
+        raise AssertionError("--keep-original did not keep the original")
+    print("  ok   --keep-original still keeps the original")
+
+
+def main():
+    work_root = tempfile.mkdtemp(prefix="steg-reveal-keep-test-")
+    try:
+        print("reveal deleting the original")
+        test_nothing_written_keeps_the_original(work_root)
+        test_normal_reveal_still_deletes_the_original(work_root)
+        test_keep_original_flag_still_works(work_root)
+    finally:
+        shutil.rmtree(work_root, ignore_errors=True)
+
+    print("reveal keep-original tests passed.")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())