diff --git a/scripts/ci/pi-intel/test_ci_helpers.py b/scripts/ci/pi-intel/test_ci_helpers.py index 300e67f721..ea01a5563f 100644 --- a/scripts/ci/pi-intel/test_ci_helpers.py +++ b/scripts/ci/pi-intel/test_ci_helpers.py @@ -1,14 +1,16 @@ import os from pathlib import Path import shutil +import subprocess import tarfile import tempfile import unittest -from unittest.mock import Mock +from unittest.mock import Mock, patch from lifecycle import CleanupUncertain, Lifecycle, run_test_cases from retain_pack import retain_pack from source_guard import tree, sha from admit_evidence import admit_evidence, regular_bytes +from verify import tool_invocation, verify_tool_invocation class CleanupRegression(unittest.TestCase): def test_first_cleanup_failure_prevents_second_test_and_scratch_removal(self): @@ -86,4 +88,24 @@ class ArtifactStorageRegression(unittest.TestCase): root=Path(root); (root/'escape').symlink_to('/etc/hosts') with self.assertRaisesRegex(RuntimeError, 'symlink'): admit_evidence(root) +class MulticallInvocationRegression(unittest.TestCase): + def test_symlink_dispatched_cli_keeps_invocation_name_and_audits_target(self): + with tempfile.TemporaryDirectory() as root: + root=Path(root); target=root/'rustup-init'; invocation=root/'rustup' + target.write_text('#!/bin/sh\ncase "$0" in */rustup) test "$1" = toolchain; exit $?;; *) exit 2;; esac\n') + target.chmod(0o755); invocation.symlink_to(target.name) + with patch('verify.shutil.which', return_value=str(invocation)): + executable,audit=tool_invocation('rustup') + self.assertEqual(executable,invocation) + self.assertEqual(audit['resolvedTarget'],str(target.resolve())) + self.assertEqual(audit['targetSha256'],sha(target)) + verify_tool_invocation(executable,audit) + clean={'PATH':'/usr/bin:/bin'} + self.assertEqual(subprocess.run([executable,'toolchain','install','1.97.1'],env=clean,timeout=5).returncode,0) + # This reproduces the CI failure: resolving the alias changes multicall mode. + self.assertEqual(subprocess.run([executable.resolve(),'toolchain','install','1.97.1'],env=clean,timeout=5).returncode,2) + target.write_text('#!/bin/sh\nexit 0\n') + with self.assertRaisesRegex(RuntimeError,'target bytes changed'): + verify_tool_invocation(executable,audit) + if __name__ == '__main__': unittest.main() diff --git a/scripts/ci/pi-intel/verify.py b/scripts/ci/pi-intel/verify.py index adfff513a9..540146452e 100644 --- a/scripts/ci/pi-intel/verify.py +++ b/scripts/ci/pi-intel/verify.py @@ -15,6 +15,19 @@ SOURCE=PIN['sourceRevision'] sha=source_guard.sha require=source_guard.require +def tool_invocation(name): + found=shutil.which(name) + require(found is not None,'Required executable not found: '+name) + # Multicall CLIs dispatch on argv[0]; preserve the name returned by PATH lookup. + invocation=Path(found).absolute() + target=invocation.resolve(strict=True) + require(invocation.name==name and target.is_file() and os.access(invocation,os.X_OK),'Unexpected executable invocation') + return invocation, {'invocationPath':str(invocation),'resolvedTarget':str(target),'targetSha256':sha(target)} + +def verify_tool_invocation(invocation, audit): + require(str(invocation)==audit['invocationPath'] and str(invocation.resolve(strict=True))==audit['resolvedTarget'], 'Executable invocation or target changed') + require(sha(invocation.resolve(strict=True))==audit['targetSha256'],'Executable target bytes changed') + def execute(args): stage=args.source.resolve(strict=True);out=args.output.resolve() out.mkdir(mode=0o700,parents=True,exist_ok=False) @@ -31,7 +44,8 @@ def execute(args): inherited_path=os.environ['PATH'] node=Path(shutil.which('node')).resolve(strict=True) pnpm=Path(shutil.which('pnpm')).resolve(strict=True) - rustup=Path(shutil.which('rustup')).resolve(strict=True) + rustup,rustup_audit=tool_invocation('rustup') + receipt['rustupTool']=rustup_audit;save() env={'PATH':inherited_path,'HOME':str(scratch/'home'),'TMPDIR':str(scratch),'LANG':'en_US.UTF-8','CI':'true', 'PAPERCLIP_TELEMETRY_ENABLED':'false','PAPERCLIP_RUNNER_SOURCE_REVISION':SOURCE, 'CARGO_HOME':str(scratch/'cargo'),'RUSTUP_HOME':str(scratch/'rustup'),'CARGO_TARGET_DIR':str(scratch/'target'), @@ -43,6 +57,7 @@ def execute(args): signal.signal(signal.SIGTERM,stop_signal) def run(command,label,timeout,cwd=stage,command_env=None,owned=False): nonlocal active,owner + if str(command[0])==str(rustup):verify_tool_invocation(rustup,rustup_audit) lifecycle.begin() row={'label':label,'startedAt':datetime.datetime.now(datetime.timezone.utc).isoformat(),'command':list(map(str,command)),'deadlineSeconds':timeout,'status':'running'} receipt['commands'].append(row);save();start=time.monotonic()