From 2e4174519e97e70f51e38bcf92273325f121f64d Mon Sep 17 00:00:00 2001 From: CL <167851968+cenglin123@users.noreply.github.com> Date: Mon, 21 Sep 2026 00:49:03 +0800 Subject: [PATCH] Keep the original stego file when a reveal extracts nothing reveal_file() removed the input whenever an extraction method reported success, without checking that anything had reached the disk. Several shapes make a method return True while writing zero files: an archive whose members are all directory entries (zipping an empty folder produces one), an empty archive, or member names that sanitize_path() rejects. The archive inside the stego file is then the only copy left, and it was deleted. The decision now rests on a before/after snapshot of the output directory (relative path -> size and mtime in ns), not on the methods' return values. The input file is excluded from the snapshot, as are the 7z_ / 7z_unzip_ temp directories _extract_with_7zip creates, whose contents are intermediates rather than results -- if its cleanup ever fails, counting them would reintroduce the deletion this guard exists to prevent. A per-call counter incremented at the four places that write extracted files was tried first and was wrong, which is worth recording: 7-Zip is invoked with -o when auto-rename is off, so a subprocess writes the results and no in-process counter sees them. That version passed the "nothing extracted" case but broke the default for a normal reveal -- the original survived a successful extraction. The snapshot covers subprocess writers as well, which is why it is the mechanism. When a method reports success and nothing appeared, the original is kept and the reason is logged, rather than falling through to the generic "all methods failed" list, which would be misleading -- a method did succeed. Verified with scripts/test-reveal-keeps-original.py, covering both halves so that a guard which simply stopped deleting anything could not pass: nothing extracted keeps the original, a normal reveal still deletes it (delete_original_after_reveal defaults to True), and --keep-original still keeps it. Mutation-tested by replacing the guard condition with "if False": the first case then fails on the deleted stego file, and removing the mutation restores a byte-identical source. Not changed, deliberately: without --auto-rename, extracting into a directory that already holds a same-named file overwrites it silently (_resolve_conflict_path returns the same path and the writer opens it with 'wb'). That is a separate behaviour with its own decision to make, and it is what made this defect look like a "nothing was extracted" case when it was first hit. --- .github/workflows/ci.yml | 1 + Steganographier.py | 45 ++++++- scripts/test-reveal-keeps-original.py | 167 ++++++++++++++++++++++++++ 3 files changed, 210 insertions(+), 3 deletions(-) create mode 100644 scripts/test-reveal-keeps-original.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 349bf54..49e256a 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/Steganographier.py b/Steganographier.py index 109da48..9b6ec35 100644 --- a/Steganographier.py +++ b/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 diff --git a/scripts/test-reveal-keeps-original.py b/scripts/test-reveal-keeps-original.py new file mode 100644 index 0000000..b638899 --- /dev/null +++ b/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())