From f94313e9d74c8ebdb1476afb6b3b2e366aadf70d Mon Sep 17 00:00:00 2001 From: mikecarper Date: Sat, 3 Oct 2026 19:31:10 -0700 Subject: [PATCH] Protect build cleanup from unsafe output paths --- .github/workflows/run-unit-tests.yml | 3 + build_legacy.sh | 55 +++++++++- test/test_build_output_safety.py | 147 +++++++++++++++++++++++++++ 3 files changed, 200 insertions(+), 5 deletions(-) create mode 100644 test/test_build_output_safety.py diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index f5511fced..aa792e763 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -91,6 +91,9 @@ jobs: - name: Run Unit Tests run: pio test -e native -e native_kiss_modem -vv + - name: Verify build output cleanup safety + run: python3 -B test/test_build_output_safety.py -v + - name: Verify serial mOTA response deadlines and RX fairness run: python3 -B test/test_mota_serial_deadline.py -v diff --git a/build_legacy.sh b/build_legacy.sh index 622183162..a6b3da9ec 100755 --- a/build_legacy.sh +++ b/build_legacy.sh @@ -5771,20 +5771,65 @@ configure_effective_build_profile() { prepare_output_dir() { local output_dir="$OUTPUT_DIR" + local canonical_output_dir + local checkout_dir - if [ -z "$output_dir" ] || [ "$output_dir" == "/" ] || [ "$output_dir" == "." ]; then - echo "Refusing to clean unsafe output directory: $output_dir" + checkout_dir=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P) || exit 1 + if ! canonical_output_dir=$(python3 - "$output_dir" "$checkout_dir" "$RESUME_BUILD_OUTPUT" <<'PY' +from pathlib import Path +import os +import subprocess +import sys + +try: + raw, checkout, resume = sys.argv[1:] + if not raw or '\n' in raw or '\r' in raw: + raise ValueError("empty or multiline output path") + requested = Path(raw) + output = requested.resolve() + checkout = Path(checkout).resolve() + protected = (Path.cwd().resolve(), checkout, Path.home().resolve()) + if any(output == path or output in path.parents for path in protected): + raise ValueError("output is a protected directory or its ancestor") + if resume != "1" and requested.is_symlink(): + raise ValueError("cleaning a symlink would replace it; use --resume or its explicit target") + if requested.exists() and not requested.is_dir(): + raise ValueError("output already exists and is not a directory") + def git_path(*arguments): + value = subprocess.check_output(["git", "-C", str(checkout), *arguments], + stderr=subprocess.PIPE).rstrip(b'\n') + return (checkout / os.fsdecode(value)).resolve() + metadata = {checkout / ".git", git_path("rev-parse", "--absolute-git-dir"), + git_path("rev-parse", "--git-common-dir")} + if any(output == path or output in path.parents or path in output.parents + for path in metadata): + raise ValueError("output overlaps Git metadata") + tracked = subprocess.check_output(["git", "-C", str(checkout), "ls-files", "-z"], + stderr=subprocess.PIPE) + if any(output == path or output in path.parents + for name in tracked.split(b'\0') if name + for path in [(checkout / os.fsdecode(name)).resolve()]): + raise ValueError("output contains tracked checkout files") + print(output) +except (OSError, RuntimeError, ValueError, subprocess.CalledProcessError) as error: + print(f"Unsafe build output path: {error}", file=sys.stderr) + sys.exit(1) +PY + ); then + echo "Refusing unsafe output directory: $output_dir" >&2 exit 1 fi if [ "$RESUME_BUILD_OUTPUT" == "1" ]; then - mkdir -p -- "$output_dir" + mkdir -p -- "$output_dir" || exit 1 echo "Resuming build output in ${output_dir}; existing artifacts will be skipped." return 0 fi - rm -rf -- "$output_dir" - mkdir -p -- "$output_dir" + # Resolve aliases before deletion, and never replace a user's output symlink. + # Parent symlinks may still point to a legitimate isolated output directory. + rm -rf -- "$canonical_output_dir" || exit 1 + mkdir -p -- "$canonical_output_dir" || exit 1 } run_resolved_build_targets() { diff --git a/test/test_build_output_safety.py b/test/test_build_output_safety.py new file mode 100644 index 000000000..fe959bee3 --- /dev/null +++ b/test/test_build_output_safety.py @@ -0,0 +1,147 @@ +"""Reject dangerous cleanup paths without touching any existing build output.""" +from pathlib import Path +import subprocess +import tempfile +import unittest + +ROOT = Path(__file__).resolve().parents[1] + + +class Tests(unittest.TestCase): + def prepare(self, output, *, cwd=ROOT, resume=False, stub=False, overrides=''): + script = r''' +source "$1/build_legacy.sh" +OUTPUT_DIR=$2 +RESUME_BUILD_OUTPUT=$3 +''' + if stub: + # Protected-path tests never execute a real filesystem mutation, + # even if this regression is reintroduced in production. + script += r''' +rm() { printf 'UNSAFE_DELETE_CALL:%s\n' "$*"; } +mkdir() { printf 'UNSAFE_MKDIR_CALL:%s\n' "$*"; } +''' + script += overrides + '\nprepare_output_dir\n' + return subprocess.run(['bash', '-c', script, 'output-safety', str(ROOT), + str(output), '1' if resume else '0'], cwd=cwd, + capture_output=True, text=True, timeout=10) + + def assert_rejected_without_mutation(self, output, **kwargs): + result = self.prepare(output, stub=True, **kwargs) + self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertIn('Refusing unsafe output directory:', result.stderr) + self.assertNotIn('UNSAFE_DELETE_CALL:', result.stdout) + self.assertNotIn('UNSAFE_MKDIR_CALL:', result.stdout) + + def test_roots_homes_checkout_ancestors_and_dot_aliases_fail_before_rm(self): + targets = ('', '/', '//', '.', './', '..', '../', str(ROOT), + str(ROOT/'out/..'), str(ROOT.parent), str(Path.home()), + *(str(parent) for parent in ROOT.parents)) + for resume in (False, True): + for output in dict.fromkeys(targets): + with self.subTest(output=output, resume=resume): + self.assert_rejected_without_mutation(output, resume=resume) + + def test_protected_symlink_targets_and_loops_are_not_deleted(self): + with tempfile.TemporaryDirectory(prefix='mesh-build-output-safety-') as temporary: + directory = Path(temporary) + for i, protected in enumerate((ROOT, ROOT.parent, Path.home(), Path('/'))): + link = directory/f'protected-{i}' + link.symlink_to(protected, target_is_directory=True) + for output in (link, str(link)+'/', str(link)+'/out/..'): + self.assert_rejected_without_mutation(output) + self.assertTrue(link.is_symlink()) + loop = directory/'loop' + loop.symlink_to(loop) + self.assert_rejected_without_mutation(loop) + self.assertTrue(loop.is_symlink()) + + def test_tracked_subtrees_and_git_metadata_fail_before_rm_even_through_parent_links(self): + for output in (ROOT/'src', ROOT/'examples', ROOT/'scripts', ROOT/'test', + ROOT/'.git', ROOT/'.git/objects', ROOT/'.git/config'): + with self.subTest(output=output): + self.assert_rejected_without_mutation(output) + with tempfile.TemporaryDirectory(prefix='mesh-build-output-safety-') as temporary: + link = Path(temporary)/'checkout-parent-link' + link.symlink_to(ROOT, target_is_directory=True) + for suffix in ('src', 'examples', '.git/objects'): + self.assert_rejected_without_mutation(link/suffix) + self.assertTrue(link.is_symlink()) + + def test_safe_output_directory_is_cleaned_and_created_without_touching_siblings(self): + with tempfile.TemporaryDirectory(prefix='mesh-build-output-safety-') as temporary: + directory = Path(temporary) + sibling = directory/'keep.bin'; sibling.write_bytes(b'keep sibling') + for relative in ('output', 'path with spaces', 'nested/output', + 'output; touch injected', '$(touch injected)'): + output = directory/relative; output.mkdir(parents=True) + (output/'old.bin').write_bytes(b'old diagnostic artifact') + result = self.prepare(relative, cwd=directory) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertTrue(output.is_dir()) + self.assertEqual(list(output.iterdir()), []) + self.assertEqual(sibling.read_bytes(), b'keep sibling') + self.assertFalse((directory/'injected').exists()) + result = self.prepare(directory/'not-created-yet'/'output') + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertTrue((directory/'not-created-yet'/'output').is_dir()) + + def test_resume_preserves_output_files_and_safe_symlink_but_clean_refuses_the_link(self): + with tempfile.TemporaryDirectory(prefix='mesh-build-output-safety-') as temporary: + directory = Path(temporary) + output = directory/'output'; output.mkdir() + artifact = output/'previous.bin'; artifact.write_bytes(b'previous diagnostic build') + initial = artifact.stat() + link = directory/'output-link'; link.symlink_to(output, target_is_directory=True) + for selected in (output, link): + result = self.prepare(selected, resume=True) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertEqual(artifact.read_bytes(), b'previous diagnostic build') + self.assertEqual((artifact.stat().st_ino, artifact.stat().st_mtime_ns), + (initial.st_ino, initial.st_mtime_ns)) + self.assert_rejected_without_mutation(link) + self.assert_rejected_without_mutation(str(link)+'/') + self.assertTrue(link.is_symlink()) + self.assertEqual(artifact.read_bytes(), b'previous diagnostic build') + + def test_safe_parent_symlink_keeps_link_and_other_outputs(self): + with tempfile.TemporaryDirectory(prefix='mesh-build-output-safety-') as temporary: + directory = Path(temporary) + physical = directory/'physical'; physical.mkdir() + output = physical/'output'; output.mkdir() + (output/'old.bin').write_bytes(b'old isolated diagnostic artifact') + sibling = physical/'keep.bin'; sibling.write_bytes(b'keep sibling') + link = directory/'parent-link'; link.symlink_to(physical, target_is_directory=True) + result = self.prepare(link/'output') + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertTrue(link.is_symlink()) + self.assertTrue(output.is_dir()) + self.assertEqual(list(output.iterdir()), []) + self.assertEqual(sibling.read_bytes(), b'keep sibling') + + def test_existing_regular_files_are_preserved_and_rejected(self): + with tempfile.TemporaryDirectory(prefix='mesh-build-output-safety-') as temporary: + path = Path(temporary)/'not-a-build-directory' + path.write_bytes(b'user data') + for resume in (False, True): + self.assert_rejected_without_mutation(path, resume=resume) + self.assertEqual(path.read_bytes(), b'user data') + + def test_delete_and_mkdir_failures_abort_before_build_can_continue(self): + with tempfile.TemporaryDirectory(prefix='mesh-build-output-safety-') as temporary: + output = Path(temporary)/'output'; output.mkdir() + (output/'previous.bin').write_bytes(b'previous isolated diagnostic artifact') + result = self.prepare(output, overrides=r''' +rm() { printf 'FAILED_DELETE\n'; return 67; } +mkdir() { printf 'SHOULD_NOT_MKDIR\n'; } +''') + self.assertNotEqual(result.returncode, 0) + self.assertIn('FAILED_DELETE', result.stdout) + self.assertNotIn('SHOULD_NOT_MKDIR', result.stdout) + self.assertTrue((output/'previous.bin').exists()) + result = self.prepare(output, resume=True, overrides='mkdir() { return 73; }') + self.assertNotEqual(result.returncode, 0) + self.assertTrue((output/'previous.bin').exists()) + +if __name__ == '__main__': + unittest.main()