| #!/usr/bin/env python3 | |
| # Copyright 2026 The Brotli Authors. All rights reserved. | |
| # | |
| # Distributed under MIT license. | |
| # See file LICENSE for detail or copy at https://opensource.org/licenses/MIT | |
| """Regression suite for brotli CLI CopyStat behavior. | |
| Usage: | |
| python3 tests/regression/t01/copystat_regression_test.py /path/to/brotli | |
| """ | |
| import argparse | |
| import contextlib | |
| import hashlib | |
| import os | |
| import shutil | |
| import stat | |
| import subprocess | |
| import tempfile | |
| import unittest | |
| from pathlib import Path | |
| PLAIN_BYTES = b"A" * 65537 | |
| TARGET_BYTES = b"TARGET\n" | |
| @contextlib.contextmanager | |
| def umask(mask): | |
| previous = os.umask(mask) | |
| try: | |
| yield | |
| finally: | |
| os.umask(previous) | |
| def file_mode(path): | |
| return stat.S_IMODE(os.stat(path).st_mode) | |
| def sha256(path): | |
| h = hashlib.sha256() | |
| with open(path, "rb") as f: | |
| for chunk in iter(lambda: f.read(65536), b""): | |
| h.update(chunk) | |
| return h.hexdigest() | |
| class CopyStatRegressionTest(unittest.TestCase): | |
| brotli = None | |
| def setUp(self): | |
| self.tmpdir = tempfile.TemporaryDirectory() | |
| self.addCleanup(self.tmpdir.cleanup) | |
| self.workdir = Path(self.tmpdir.name) | |
| def write_plain(self, path): | |
| path.write_bytes(PLAIN_BYTES) | |
| def write_target(self, path): | |
| path.write_bytes(TARGET_BYTES) | |
| def on_subprocess_failed(self, preamble: str, proc): | |
| retcode = proc.returncode | |
| stdout = proc.stdout.decode("utf-8", "replace") | |
| stderr = proc.stderr.decode("utf-8", "replace") | |
| self.fail( | |
| f"{preamble}\n" | |
| f"retcode:{retcode}\n" | |
| f"stdout:\n{stdout}\n" | |
| f"stderr:\n{stderr}\n") | |
| def run_brotli(self, *args, input_bytes=None, env=None, check=True): | |
| proc = subprocess.run( | |
| [str(self.brotli)] + [str(arg) for arg in args], | |
| input=input_bytes, | |
| capture_output=True, | |
| env=env, | |
| check=False) | |
| if check and proc.returncode != 0: | |
| self.on_subprocess_failed("brotli failed", proc) | |
| return proc | |
| def compress(self, src, dst, no_copy_stat=False): | |
| args = ["-f", "-k"] | |
| if no_copy_stat: | |
| args.append("-n") | |
| args.extend([src, "-o", dst]) | |
| return self.run_brotli(*args) | |
| def decompress(self, src, dst, no_copy_stat=False): | |
| args = ["-d", "-f"] | |
| if no_copy_stat: | |
| args.append("-n") | |
| args.extend([src, "-o", dst]) | |
| return self.run_brotli(*args) | |
| def test_modes_propagate_on_compress_and_decompress(self): | |
| modes = [0o600, 0o640, 0o644, 0o664, 0o660, | |
| 0o400, 0o444, 0o755, 0o700, 0o604] | |
| with umask(0o022): | |
| for mode in modes: | |
| with self.subTest(mode=f"{mode:03o}"): | |
| case_dir = self.workdir / (f"{mode:03o}") | |
| case_dir.mkdir() | |
| src = case_dir / "in.bin" | |
| compressed = case_dir / "in.bin.br" | |
| decoded = case_dir / "out.bin" | |
| self.write_plain(src) | |
| os.chmod(src, mode) | |
| self.compress(src, compressed) | |
| self.decompress(compressed, decoded) | |
| self.assertEqual(file_mode(compressed), mode) | |
| self.assertEqual(file_mode(decoded), mode) | |
| self.assertEqual(decoded.read_bytes(), PLAIN_BYTES) | |
| def test_special_mode_bits_are_masked(self): | |
| with umask(0o022): | |
| src = self.workdir / "in.bin" | |
| compressed = self.workdir / "in.bin.br" | |
| self.write_plain(src) | |
| os.chmod(src, 0o3644) | |
| actual_input_mode = file_mode(src) | |
| if actual_input_mode == 0o644: | |
| self.skipTest("filesystem stripped special mode bits from input") | |
| self.compress(src, compressed) | |
| compressed_mode = file_mode(compressed) | |
| self.assertEqual(compressed_mode & 0o7000, 0) | |
| self.assertEqual(compressed_mode & 0o777, actual_input_mode & 0o777) | |
| def test_no_copy_stat_flag_suppresses_mode_copy(self): | |
| with umask(0o022): | |
| src = self.workdir / "in.bin" | |
| without_copy = self.workdir / "without_copy.br" | |
| with_copy = self.workdir / "with_copy.br" | |
| self.write_plain(src) | |
| os.chmod(src, 0o606) | |
| self.compress(src, without_copy, no_copy_stat=True) | |
| self.compress(src, with_copy) | |
| self.assertNotEqual(file_mode(without_copy), 0o606) | |
| self.assertEqual(file_mode(with_copy), 0o606) | |
| def test_timestamp_is_copied(self): | |
| with umask(0o022): | |
| src = self.workdir / "in.bin" | |
| compressed = self.workdir / "in.bin.br" | |
| decoded = self.workdir / "out.bin" | |
| self.write_plain(src) | |
| expected_mtime = 1577934245 | |
| os.utime(src, (expected_mtime, expected_mtime)) | |
| self.compress(src, compressed) | |
| self.decompress(compressed, decoded) | |
| self.assertEqual(int(os.stat(compressed).st_mtime), expected_mtime) | |
| self.assertEqual(int(os.stat(decoded).st_mtime), expected_mtime) | |
| def test_stdin_input_has_no_path_to_copy_from(self): | |
| payload = b"stdin -> compressed file\n" | |
| compressed = self.run_brotli("-c", input_bytes=payload).stdout | |
| decoded = self.workdir / "decoded.txt" | |
| self.run_brotli("-d", "-o", decoded, input_bytes=compressed) | |
| self.assertEqual(decoded.read_bytes(), payload) | |
| self.assertEqual(file_mode(decoded) & ~0o666, 0) | |
| def test_stdout_path_skips_copystat(self): | |
| payload = b"hello brotli stdout pipeline\n" | |
| compressed = self.run_brotli("-c", input_bytes=payload).stdout | |
| decoded = self.run_brotli("-d", "-c", input_bytes=compressed).stdout | |
| self.assertEqual(decoded, payload) | |
| def test_content_roundtrip(self): | |
| cases = { | |
| "small-ascii": b"hello brotli\n", | |
| "empty": b"", | |
| "plain-65537": PLAIN_BYTES, | |
| "aligned-65536": b"\0" * 65536, | |
| "deterministic-binary": bytes((i * 73 + 41) % 256 | |
| for i in range(12288)), | |
| } | |
| with umask(0o022): | |
| for label, data in cases.items(): | |
| with self.subTest(label=label): | |
| case_dir = self.workdir / label | |
| case_dir.mkdir() | |
| src = case_dir / "in.bin" | |
| compressed = case_dir / "in.bin.br" | |
| decoded = case_dir / "out.bin" | |
| src.write_bytes(data) | |
| os.chmod(src, 0o644) | |
| self.compress(src, compressed) | |
| if data: | |
| self.assertGreater(compressed.stat().st_size, 0) | |
| self.decompress(compressed, decoded) | |
| self.assertEqual(decoded.read_bytes(), data) | |
| def test_fclose_swap_does_not_redirect_copystat(self): | |
| compressed, out_path, target_path = self.prepare_attack_files() | |
| pre_hash = sha256(target_path) | |
| self.run_brotli_with_swap(compressed, out_path, target_path) | |
| self.assertTrue(out_path.is_symlink()) | |
| self.assertEqual(file_mode(target_path), 0o600) | |
| self.assertEqual(sha256(target_path), pre_hash) | |
| def test_fclose_swap_with_no_copy_stat_leaves_target_unchanged(self): | |
| compressed, out_path, target_path = self.prepare_attack_files() | |
| pre_hash = sha256(target_path) | |
| self.run_brotli_with_swap(compressed, out_path, target_path, | |
| no_copy_stat=True) | |
| self.assertTrue(out_path.is_symlink()) | |
| self.assertEqual(file_mode(target_path), 0o600) | |
| self.assertEqual(sha256(target_path), pre_hash) | |
| def prepare_attack_files(self): | |
| src = self.workdir / "plain.bin" | |
| compressed = self.workdir / "plain.bin.br" | |
| out_path = self.workdir / "out" | |
| target_path = self.workdir / "target.txt" | |
| self.write_plain(src) | |
| self.write_target(target_path) | |
| os.chmod(src, 0o644) | |
| os.chmod(target_path, 0o600) | |
| self.compress(src, compressed) | |
| os.chmod(compressed, 0o644) | |
| return compressed, out_path, target_path | |
| def run_brotli_with_swap(self, compressed, out_path, target_path, | |
| no_copy_stat=False): | |
| env = dict(os.environ) | |
| env["BROTLI_SWAP_OUTPUT_ABS"] = str(out_path) | |
| env["BROTLI_SWAP_TARGET_ABS"] = str(target_path) | |
| env["LD_PRELOAD"] = self.ld_preload_value() | |
| args = ["-d", "-f"] | |
| if no_copy_stat: | |
| args.append("-n") | |
| args.extend(["-o", out_path, compressed]) | |
| proc = self.run_brotli(*args, env=env, check=False) | |
| if proc.returncode != 0: | |
| self.on_subprocess_failed("brotli with fclose_swap failed", proc) | |
| def ld_preload_value(self): | |
| preloads = [] | |
| asan_runtime = self.asan_runtime() | |
| if asan_runtime: | |
| preloads.append(asan_runtime) | |
| preloads.append(str(self.swap_helper())) | |
| existing = os.environ.get("LD_PRELOAD") | |
| if existing: | |
| preloads.append(existing) | |
| return ":".join(preloads) | |
| def swap_helper(self): | |
| if not Path("/proc/self/fd").is_dir(): | |
| self.skipTest("fclose_swap helper requires /proc/self/fd") | |
| cc = shutil.which("cc") | |
| if cc is None: | |
| self.skipTest("cc is required to build fclose_swap.so") | |
| helper_src = Path(__file__).with_name("fclose_swap.c") | |
| helper_out = self.workdir / "libfclose_swap.so" | |
| proc = subprocess.run( | |
| [cc, "-shared", "-fPIC", str(helper_src), "-o", str(helper_out), | |
| "-ldl"], | |
| capture_output=True, | |
| check=False) | |
| if proc.returncode != 0: | |
| self.on_subprocess_failed("failed to build fclose_swap.so", proc) | |
| return helper_out | |
| def asan_runtime(self): | |
| try: | |
| proc = subprocess.run(["ldd", str(self.brotli)], | |
| capture_output=True, | |
| check=False) | |
| except FileNotFoundError: | |
| return None | |
| if b"libasan" not in proc.stdout + proc.stderr: | |
| return None | |
| cc = shutil.which("cc") | |
| if cc is None: | |
| return None | |
| proc = subprocess.run([cc, "-print-file-name=libasan.so"], | |
| capture_output=True, | |
| text=True, | |
| check=False) | |
| candidate = proc.stdout.strip() | |
| if candidate and Path(candidate).is_file(): | |
| return candidate | |
| return None | |
| def main(): | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("brotli", help="path to the brotli CLI binary to test") | |
| parser.add_argument("-v", "--verbose", action="store_true") | |
| args = parser.parse_args() | |
| brotli = Path(args.brotli).resolve() | |
| if not brotli.is_file(): | |
| parser.error(f"{brotli} is not a file") | |
| CopyStatRegressionTest.brotli = brotli | |
| unittest.main(argv=[__file__], verbosity=2 if args.verbose else 1) | |
| if __name__ == "__main__": | |
| main() |