98 lines
4.5 KiB
Python
98 lines
4.5 KiB
Python
"""Windows-only regressions for reviewed firmware and process-tree risks."""
|
|
import ctypes
|
|
import hashlib
|
|
import importlib.util
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
import device_adapter as adapter
|
|
import tool_runner as runner
|
|
|
|
|
|
def process_exited(pid):
|
|
k = ctypes.WinDLL("kernel32", use_last_error=True)
|
|
k.OpenProcess.argtypes = [ctypes.c_uint, ctypes.c_int, ctypes.c_uint]
|
|
k.OpenProcess.restype = ctypes.c_void_p
|
|
k.WaitForSingleObject.argtypes = [ctypes.c_void_p, ctypes.c_uint]
|
|
k.CloseHandle.argtypes = [ctypes.c_void_p]
|
|
handle = k.OpenProcess(0x100000, False, pid)
|
|
if not handle:
|
|
return ctypes.get_last_error() == 87
|
|
try:
|
|
return k.WaitForSingleObject(handle, 5000) == 0
|
|
finally:
|
|
k.CloseHandle(handle)
|
|
|
|
|
|
@unittest.skipUnless(os.name == "nt", "Windows lock and job regression")
|
|
class ReviewRegression(unittest.TestCase):
|
|
def test_snapshot_is_bound_and_blocks_modification_and_replacement(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
original = Path(d) / "source.rom"
|
|
for replacement in (b"B" * 64, b"B" * 10, b"B" * 128):
|
|
original.write_bytes(b"A" * 64)
|
|
digest = hashlib.sha256(original.read_bytes()).hexdigest()
|
|
with adapter.protected_rom_snapshot(original, digest, 64) as snapshot:
|
|
original.write_bytes(replacement)
|
|
with self.assertRaises(OSError):
|
|
snapshot.write_bytes(replacement)
|
|
with self.assertRaises(OSError):
|
|
snapshot.unlink()
|
|
self.assertEqual(snapshot.read_bytes(), b"A" * 64)
|
|
self.assertFalse(snapshot.exists())
|
|
|
|
def test_snapshot_rejects_changed_source_before_transfer(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
original = Path(d) / "source.rom"
|
|
original.write_bytes(b"B" * 64)
|
|
digest = hashlib.sha256(b"A" * 64).hexdigest()
|
|
with self.assertRaises(adapter.protocol.UpgradeError):
|
|
with adapter.protected_rom_snapshot(original, digest, 64):
|
|
self.fail("modified image must never be passed to transfer")
|
|
|
|
def test_timeout_reaps_descendant(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
pidfile = Path(d) / "pid"
|
|
code = ("import subprocess,sys,time,pathlib; "
|
|
"p=subprocess.Popen([sys.executable,'-c','import time; time.sleep(90)']); "
|
|
"pathlib.Path(sys.argv[1]).write_text(str(p.pid)); time.sleep(90)")
|
|
with self.assertRaises(subprocess.TimeoutExpired):
|
|
runner._run_in_windows_job([sys.executable, "-c", code, str(pidfile)],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=2)
|
|
self.assertTrue(pidfile.exists())
|
|
self.assertTrue(process_exited(int(pidfile.read_text())))
|
|
|
|
def test_worker_termination_closes_job_and_reaps_descendant(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
pidfile = Path(d) / "pid"
|
|
child_code = ("import subprocess,sys,time,pathlib; "
|
|
"p=subprocess.Popen([sys.executable,'-c','import time; time.sleep(90)']); "
|
|
"pathlib.Path(sys.argv[1]).write_text(str(p.pid)); time.sleep(90)")
|
|
worker_code = ("import sys; sys.path.insert(0,sys.argv[1]); import tool_runner; "
|
|
"tool_runner._run_in_windows_job([sys.executable,'-c',sys.argv[2],sys.argv[3]],timeout=80)")
|
|
worker = subprocess.Popen([sys.executable, "-c", worker_code, str(ROOT / "scripts"), child_code, str(pidfile)],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
try:
|
|
end = time.monotonic() + 10
|
|
while not pidfile.exists() and time.monotonic() < end:
|
|
time.sleep(.05)
|
|
self.assertTrue(pidfile.exists())
|
|
pid = int(pidfile.read_text())
|
|
worker.kill()
|
|
worker.wait(timeout=5)
|
|
self.assertTrue(process_exited(pid))
|
|
finally:
|
|
if worker.poll() is None:
|
|
worker.kill()
|
|
worker.wait()
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|