Add guarded hardware acceptance and restoration tools
This commit is contained in:
mikecarper committed
1 parent
48bad8ddfb
commit
89dcd82089
18 files changed
+4529
-18
No files matched your search
@@ -0,0 +1,176 @@
|
||||
"""Offline release identity gates; no serial port, SDK, browser or radio."""
|
||||
import contextlib
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
import struct
|
||||
import sys
|
||||
import tempfile
|
||||
from types import ModuleType, SimpleNamespace
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
HIL = Path(__file__).resolve().parents[1] / "tools/hil"
|
||||
sys.path.insert(0, str(HIL))
|
||||
# The acceptance code's serial entry points are never called by this fixture.
|
||||
# Keep the offline contract runnable without installing a hardware dependency.
|
||||
serial = ModuleType("serial")
|
||||
serial.tools = ModuleType("serial.tools")
|
||||
serial.tools.list_ports = SimpleNamespace(comports=lambda: [])
|
||||
with mock.patch.dict(sys.modules, {"serial": serial, "serial.tools": serial.tools}):
|
||||
spec = importlib.util.spec_from_file_location("companion_acceptance", HIL / "companion_release_acceptance.py")
|
||||
acceptance = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(acceptance)
|
||||
|
||||
VERSION = "v1.18.0.1-halo-keymind-cascade-dev-652289ee"
|
||||
BUILD = "18-Oct-2026"
|
||||
FULL_VERSION = VERSION + " (Build: " + BUILD + ")"
|
||||
|
||||
|
||||
def fixed(value, size):
|
||||
return value.encode("ascii").ljust(size, b"\0")
|
||||
|
||||
|
||||
def self_info():
|
||||
frame = bytearray(58)
|
||||
frame[:4] = bytes((5, 1, 20, 22))
|
||||
frame[4:36] = bytes(range(1, 33))
|
||||
struct.pack_into("<II", frame, 48, 910525, 62500)
|
||||
frame[56:58] = bytes((7, 5))
|
||||
return bytes(frame)
|
||||
|
||||
|
||||
def device_info(board="Heltec V4"):
|
||||
return (bytes((13, 14, 50, 1)) + struct.pack("<I", 987654)
|
||||
+ fixed(BUILD, 12) + fixed(board, 40) + fixed(VERSION[:19], 20)
|
||||
+ bytes((0, 2)))
|
||||
|
||||
|
||||
class WireSession(acceptance.Session):
|
||||
"""Exercise production handshake/request validation with inert frames."""
|
||||
version_reply = b"\x1d7f|" + FULL_VERSION.encode("ascii")
|
||||
board_reply = device_info()
|
||||
sent = []
|
||||
|
||||
def __enter__(self):
|
||||
self.handshake()
|
||||
return self
|
||||
|
||||
def __exit__(self, *_):
|
||||
pass
|
||||
|
||||
def send(self, payload):
|
||||
self.sent.append(payload)
|
||||
self.pending = payload
|
||||
|
||||
def receive(self, timeout=None):
|
||||
if self.pending[:1] == b"\x01":
|
||||
return self_info()
|
||||
if self.pending[:1] == b"\x16":
|
||||
return self.board_reply
|
||||
if self.pending[:1] == b"\x42":
|
||||
return self.version_reply
|
||||
raise AssertionError("Unexpected request; no mutation is allowed by this wire fixture")
|
||||
|
||||
def contacts(self):
|
||||
return {}
|
||||
|
||||
def channels(self):
|
||||
return [{"index": 0, "name_hex": "70726976617465", "key_hex": "ab" * 32}]
|
||||
|
||||
def stats(self):
|
||||
return {"uptime_seconds": 100, "error_flags": 0}
|
||||
|
||||
|
||||
class ReleaseIdentityTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
WireSession.sent = []
|
||||
WireSession.version_reply = b"\x1d7f|" + FULL_VERSION.encode("ascii")
|
||||
WireSession.board_reply = device_info()
|
||||
|
||||
def session(self):
|
||||
return WireSession(SimpleNamespace(expected_version=VERSION, expected_board="Heltec V4", serial="ExactUSB123"),
|
||||
acceptance.TransportCounters())
|
||||
|
||||
def test_exact_complete_candidate_and_device_build_are_verified(self):
|
||||
session = self.session()
|
||||
snapshot = session.snapshot()
|
||||
self.assertEqual(WireSession.sent[-1], b"\x427f|ver")
|
||||
self.assertEqual(session.full_cli_version, FULL_VERSION)
|
||||
self.assertEqual(snapshot["full_cli_version"], FULL_VERSION)
|
||||
self.assertEqual(snapshot["full_cli_version_frame_hex"], WireSession.version_reply.hex())
|
||||
self.assertEqual(acceptance.stable_state(snapshot)["full_cli_version"], FULL_VERSION)
|
||||
|
||||
def test_same_19_byte_prefix_wrong_source_is_rejected_before_crud(self):
|
||||
other = VERSION.replace("652289ee", "d3e05abc")
|
||||
self.assertEqual(other[:19], VERSION[:19])
|
||||
WireSession.version_reply = b"\x1d7f|" + (other + " (Build: " + BUILD + ")").encode()
|
||||
with tempfile.TemporaryDirectory() as temporary:
|
||||
directory = Path(temporary)
|
||||
args = SimpleNamespace(mode="owned-crud", expected_version=VERSION, expected_board="Heltec V4",
|
||||
manifest=None, serial="ExactUSB123", snapshot=directory / "private.json",
|
||||
report=directory / "report.json")
|
||||
output = io.StringIO()
|
||||
with mock.patch.object(acceptance, "parse_args", return_value=args), \
|
||||
mock.patch.object(acceptance, "Session", WireSession), \
|
||||
mock.patch.object(acceptance, "owned_plan") as plan, \
|
||||
mock.patch.object(acceptance, "owned_crud") as crud, \
|
||||
contextlib.redirect_stdout(output):
|
||||
self.assertEqual(acceptance.main(), 1)
|
||||
plan.assert_not_called()
|
||||
crud.assert_not_called()
|
||||
self.assertFalse(args.snapshot.exists())
|
||||
report = json.loads(args.report.read_text())
|
||||
self.assertFalse(report["passed"])
|
||||
self.assertFalse(report["source_version_verified"])
|
||||
self.assertNotIn("full_runtime_version", report)
|
||||
self.assertNotIn(WireSession.board_reply.hex(), output.getvalue())
|
||||
|
||||
def test_malformed_or_uncorrelated_cli_reply_is_rejected(self):
|
||||
for reply in (b"\x1d" + FULL_VERSION.encode(), b"\x1d7g|" + FULL_VERSION.encode(),
|
||||
b"\x1d7f|" + FULL_VERSION.encode() + b"\0", b"\x1d7f|\xff"):
|
||||
with self.subTest(reply=reply[:4]):
|
||||
WireSession.version_reply = reply
|
||||
with self.assertRaises(acceptance.AcceptanceError):
|
||||
self.session().handshake()
|
||||
|
||||
def test_wrong_board_is_rejected_before_local_cli(self):
|
||||
WireSession.board_reply = device_info("Unrelated board")
|
||||
with self.assertRaisesRegex(acceptance.AcceptanceError, "board"):
|
||||
self.session().handshake()
|
||||
self.assertFalse(any(frame[0] == 0x42 for frame in WireSession.sent))
|
||||
|
||||
def test_cli_error_and_incomplete_build_formatter_are_rejected(self):
|
||||
for reply in (b"\x01\x03", b"\x1d7f|Error: unavailable",
|
||||
b"\x1d7f|" + VERSION.encode(),
|
||||
b"\x1d7f|" + (VERSION + " (Build: different)").encode()):
|
||||
with self.subTest(reply=reply[:4]):
|
||||
WireSession.version_reply = reply
|
||||
with self.assertRaises(acceptance.AcceptanceError):
|
||||
self.session().handshake()
|
||||
|
||||
def test_sanitized_pass_records_full_version_without_pin_or_channel_key(self):
|
||||
with tempfile.TemporaryDirectory() as temporary:
|
||||
directory = Path(temporary)
|
||||
args = SimpleNamespace(mode="snapshot", expected_version=VERSION, expected_board="Heltec V4",
|
||||
manifest=None, serial="ExactUSB123", snapshot=directory / "private.json",
|
||||
report=directory / "report.json")
|
||||
output = io.StringIO()
|
||||
with mock.patch.object(acceptance, "parse_args", return_value=args), \
|
||||
mock.patch.object(acceptance, "Session", WireSession), contextlib.redirect_stdout(output):
|
||||
self.assertEqual(acceptance.main(), 0)
|
||||
report = json.loads(args.report.read_text())
|
||||
self.assertTrue(report["passed"] and report["source_version_verified"])
|
||||
self.assertEqual(report["full_runtime_version"], FULL_VERSION)
|
||||
self.assertFalse(report["live_flash_hash_verified"])
|
||||
self.assertFalse(report["cold_boot_tested"])
|
||||
self.assertEqual(args.snapshot.stat().st_mode & 0o777, 0o600)
|
||||
for sanitized in (output.getvalue(), args.report.read_text()):
|
||||
self.assertNotIn("ab" * 32, sanitized)
|
||||
self.assertNotIn(WireSession.board_reply.hex(), sanitized)
|
||||
self.assertNotIn("987654", sanitized)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -20,6 +20,8 @@ import ethernet_primary_probe as primary
|
||||
|
||||
class PortOwnershipTests(unittest.TestCase):
|
||||
def run_inspector(self, result):
|
||||
result.stdout = ""
|
||||
result.stderr = ""
|
||||
return patch.object(session.subprocess, "run", return_value=result)
|
||||
|
||||
def test_existing_owner_refuses_without_killing_anything(self):
|
||||
@@ -28,7 +30,7 @@ class PortOwnershipTests(unittest.TestCase):
|
||||
self.run_inspector(SimpleNamespace(returncode=0)) as run:
|
||||
with self.assertRaisesRegex(RuntimeError, "already in use"):
|
||||
session.assert_port_idle("/dev/ttyUSB0")
|
||||
self.assertEqual(run.call_args.args[0][:2], ["/usr/bin/fuser", "--"])
|
||||
self.assertEqual(run.call_args.args[0], ["/usr/bin/fuser", "/dev/ttyUSB0"])
|
||||
self.assertNotIn("-k", run.call_args.args[0])
|
||||
|
||||
def test_idle_port_is_allowed(self):
|
||||
|
||||
@@ -0,0 +1,235 @@
|
||||
"""Offline app-only flash contracts; every flash operation is a temporary file.
|
||||
|
||||
Reuse the data-restoration test's pinned fake parser/flasher boundary. The real
|
||||
adapter runs unchanged, including input validation, guards, sole app write,
|
||||
readbacks and runtime finish. No installed flasher, serial port or PIO is used.
|
||||
"""
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import struct
|
||||
import unittest
|
||||
import zlib
|
||||
|
||||
import test_mcfirmware_bench_data_restore as fixture
|
||||
|
||||
TOOL = fixture.ROOT / 'tools/hil/mcfirmware_bench_app_flash.sh'
|
||||
IMAGE_PARSER = r'''
|
||||
def image_header(image):
|
||||
if len(image)<36 or image[0]!=0xe9: raise ExpansionError('bad image header')
|
||||
return struct.unpack_from('<H',image,12)[0],1<<(20+(image[3]>>4))
|
||||
def endf_identity(image):
|
||||
trailer=image[-56:]
|
||||
if len(trailer)!=56 or trailer[:4]!=b'EndF' or struct.unpack_from('<I',trailer,4)[0]!=len(image)-56:
|
||||
raise ExpansionError('bad EndF length')
|
||||
if trailer[8:16]!=hashlib.sha256(image[:-56]).digest()[:8]: raise ExpansionError('bad EndF hash')
|
||||
version,target=struct.unpack_from('<II',trailer,16)
|
||||
hardware=trailer[24:].rstrip(b'\0').decode('ascii')
|
||||
if not version or not target or not hardware: raise ExpansionError('bad EndF identity')
|
||||
return version,target,hardware
|
||||
'''
|
||||
|
||||
|
||||
def image(capacity, *, target=0, hardware='MOCK_BOARD', chip=9, mode=2, length=256):
|
||||
body = bytearray(length - 56)
|
||||
body[0:4] = bytes([0xe9, 1, mode, (capacity.bit_length() - 21) << 4])
|
||||
struct.pack_into('<H', body, 12, chip)
|
||||
struct.pack_into('<I', body, 32, 0xabcd5432)
|
||||
return (bytes(body) + b'EndF' + struct.pack('<I', len(body)) + hashlib.sha256(body).digest()[:8]
|
||||
+ struct.pack('<II', 0x01120001, target) + hardware.encode().ljust(32, b'\0'))
|
||||
|
||||
|
||||
def ota(slot):
|
||||
data = bytearray(b'\xff' * 8192)
|
||||
sequence = slot + 1
|
||||
struct.pack_into('<I', data, 0, sequence)
|
||||
struct.pack_into('<I', data, 28, zlib.crc32(struct.pack('<I', sequence), 0xffffffff) & 0xffffffff)
|
||||
return bytes(data)
|
||||
|
||||
|
||||
@unittest.skipUnless(os.name == 'posix' and all(shutil.which(c) for c in
|
||||
('bash', 'python3', 'awk', 'stat', 'sha256sum', 'realpath', 'timeout')),
|
||||
'POSIX shell tools required; no physical device required')
|
||||
class AppFlashTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
# This creates only private temporary files and PATH command stubs.
|
||||
fixture.DataRestoreTests.setUp(self)
|
||||
helper = self.flasher.parent / 'tools/meshcore_partition_expansion.py'
|
||||
helper.write_text(fixture.PARSER + IMAGE_PARSER)
|
||||
source = self.flasher.read_text().replace(fixture.digest(fixture.PARSER.encode()),
|
||||
fixture.digest(helper.read_bytes()))
|
||||
source = source.replace('Detected flash size:8MB\\n\' "${MOCK_CHIP:-ESP32-S3}"',
|
||||
'Detected flash size:%sMB\\n\' "${MOCK_CHIP:-ESP32-S3}" "${MOCK_MB:-8}"')
|
||||
source = source.replace("'nvs-readback'", "'application-readback'")
|
||||
source = source.replace("int(sys.argv[2],0)==0x9000 and os.environ.get('MOCK_MUTATE_AFTER_WRITE')",
|
||||
"os.environ.get('MOCK_MUTATE_AFTER_WRITE')")
|
||||
self.flasher.write_text(source)
|
||||
self.image_path = self.root / 'application.bin'
|
||||
self.configure()
|
||||
|
||||
def configure(self, *, layout='xiao', active=0, relocated=False):
|
||||
self.capacity = 0x1000000 if layout == 'v4' else 0x800000
|
||||
if layout == 'v4':
|
||||
self.apps = [(0x10000, 0x640000), (0x650000, 0x640000)]
|
||||
data = [('spiffs', 1, 0x82, 0xc90000, 0x360000), ('coredump', 1, 3, 0xff0000, 0x10000)]
|
||||
elif layout == 'indicator':
|
||||
self.apps = [(0x10000, 0x280000), (0x400000, 0x280000)]
|
||||
data = [('spiffs', 1, 0x82, 0x290000, 0x160000), ('coredump', 1, 3, 0x3f0000, 0x10000)]
|
||||
else:
|
||||
self.apps = [(0x10000, 0x330000), (0x340000, 0x330000)]
|
||||
data = [('spiffs', 1, 0x82, 0x670000, 0x190000)]
|
||||
self.active = active
|
||||
self.ota_address = 0xc000 if relocated else 0xe000
|
||||
entries = [('nvs', 1, 2, 0x9000, 0x3000 if relocated else 0x5000),
|
||||
('otadata', 1, 0, self.ota_address, 0x2000)]
|
||||
entries += [('app' + str(i), 0, 0x10 + i, *part) for i, part in enumerate(self.apps)] + data
|
||||
self.table = b''.join(struct.pack('<2sBBII16sI', b'\xaa\x50', kind, sub, offset, length,
|
||||
name.encode(), 0)
|
||||
for name, kind, sub, offset, length in entries).ljust(4096, b'\xff')
|
||||
before = bytearray(b'\xff' * self.capacity)
|
||||
before[0x8000:0x9000] = self.table
|
||||
before[self.ota_address:self.ota_address + 8192] = ota(active)
|
||||
self.before = bytes(before)
|
||||
self.backup.write_bytes(before)
|
||||
self.flash.write_bytes(before)
|
||||
self.trace.unlink(missing_ok=True)
|
||||
self.manifest.write_text(json.dumps(dict(complete=True, operation='read-only-flash-backup',
|
||||
chip='ESP32-S3', device_mac=fixture.MAC, offset=0, length=self.capacity,
|
||||
flash_bytes=self.capacity, sha256=fixture.digest(before), artifact=self.backup.name)))
|
||||
self.image = image(self.capacity)
|
||||
self.image_path.write_bytes(self.image)
|
||||
address, size = self.apps[active]
|
||||
self.args = ['bash', str(TOOL), '--port', '/dev/ttyACM99', '--usb-path', 'exact-path',
|
||||
'--usb-id', 'exact-id', '--device', 'MOCK', '--bench-dir', str(self.bench),
|
||||
'--expect-mac', fixture.MAC, '--backup', str(self.backup), '--backup-manifest', str(self.manifest),
|
||||
'--backup-sha256', fixture.digest(before), '--expect-partition-sha256', fixture.digest(self.table),
|
||||
'--address', hex(address), '--slot-size', hex(size), '--image', str(self.image_path),
|
||||
'--image-sha256', fixture.digest(self.image), '--expect-target-id', '0',
|
||||
'--expect-hardware-id', 'MOCK_BOARD', '--allow-write', '--flasher', str(self.flasher),
|
||||
'--flasher-sha256', fixture.digest(self.flasher.read_bytes())]
|
||||
|
||||
def argument(self, option, value):
|
||||
self.args[self.args.index(option) + 1] = str(value)
|
||||
|
||||
def run_tool(self, **env):
|
||||
self.trace.unlink(missing_ok=True)
|
||||
return fixture.DataRestoreTests.run_tool(self, MOCK_MB=str(self.capacity // 0x100000), **env)
|
||||
|
||||
def refuse(self, *, before_prepare=False, **env):
|
||||
result = self.run_tool(**env)
|
||||
fixture.DataRestoreTests.assert_no_writes(self, result)
|
||||
self.assertEqual(self.flash.read_bytes(), self.before)
|
||||
if before_prepare:
|
||||
self.assertNotIn('prepare', self.actions)
|
||||
|
||||
def assert_success(self, **env):
|
||||
result = self.run_tool(**env)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
writes = [line.split() for line in self.actions if 'write-flash' in line]
|
||||
self.assertEqual([int(line[6]) for line in writes], [self.apps[self.active][0]])
|
||||
expected = bytearray(self.before)
|
||||
offset = self.apps[self.active][0]
|
||||
expected[offset:offset + len(self.image)] = self.image
|
||||
self.assertEqual(self.flash.read_bytes(), expected)
|
||||
self.assertEqual(self.actions[-1], 'finish')
|
||||
self.assertFalse(any('erase' in line for line in self.actions))
|
||||
return result
|
||||
|
||||
def test_only_active_app_written_in_both_slots_of_three_real_layout_shapes(self):
|
||||
for layout in ('v4', 'xiao', 'indicator'):
|
||||
for active in (0, 1):
|
||||
with self.subTest(layout=layout, active=active):
|
||||
self.configure(layout=layout, active=active)
|
||||
self.assert_success()
|
||||
self.assertFalse(any('hard-reset' in action for action in self.actions))
|
||||
|
||||
def test_relocated_ota_read_and_uart_final_probe_reset_then_finish(self):
|
||||
self.configure(layout='indicator', active=1, relocated=True)
|
||||
self.assert_success(MOCK_AFTER='hard-reset')
|
||||
self.assertIn('--after hard-reset --baud 115200 read-mac', self.actions[-2])
|
||||
reads = [int(line.split()[6], 0) for line in self.actions if 'read-flash' in line]
|
||||
self.assertEqual(reads.count(0xc000), 2)
|
||||
self.assertNotIn(0xe000, reads)
|
||||
|
||||
def test_live_chip_or_capacity_mismatch_refused_before_write(self):
|
||||
self.refuse(MOCK_CHIP='ESP32-C3')
|
||||
# Keep the image/backup capacity at8MiB, change only the live probe.
|
||||
self.capacity = 0x1000000
|
||||
self.refuse()
|
||||
|
||||
def test_changed_live_table_or_active_ota_refused_before_write(self):
|
||||
for address, replacement in ((0x8000, b'\x00'), (self.ota_address, ota(1))):
|
||||
with self.subTest(address=address):
|
||||
changed = bytearray(self.before)
|
||||
changed[address:address + len(replacement)] = replacement
|
||||
self.flash.write_bytes(changed)
|
||||
# The altered live state itself must also remain untouched.
|
||||
self.before = bytes(changed)
|
||||
self.refuse()
|
||||
self.configure()
|
||||
|
||||
def test_wrong_requested_active_slot_or_size_refused_before_prepare(self):
|
||||
self.argument('--address', hex(self.apps[1][0]))
|
||||
self.refuse(before_prepare=True)
|
||||
self.configure()
|
||||
self.argument('--slot-size', self.apps[0][1] - 4096)
|
||||
self.refuse(before_prepare=True)
|
||||
|
||||
def test_image_hash_backup_hash_and_mac_refused_before_prepare(self):
|
||||
for option, wrong in (('--image-sha256', '0' * 64), ('--backup-sha256', '0' * 64),
|
||||
('--expect-mac', '112233445566'), ('--expect-partition-sha256', '0' * 64)):
|
||||
with self.subTest(option=option):
|
||||
self.configure()
|
||||
self.argument(option, wrong)
|
||||
self.refuse(before_prepare=True)
|
||||
|
||||
def test_full_zero_target_requires_exact_optin_and_hardware(self):
|
||||
for option, wrong in (('--expect-target-id', '1234'), ('--expect-hardware-id', 'OTHER_BOARD')):
|
||||
with self.subTest(option=option):
|
||||
self.configure()
|
||||
self.argument(option, wrong)
|
||||
self.refuse(before_prepare=True)
|
||||
|
||||
def test_endf_body_corruption_and_oversize_image_refused_before_prepare(self):
|
||||
for changed in (self.image[:40] + bytes([self.image[40] ^ 1]) + self.image[41:],
|
||||
image(self.capacity, length=self.apps[0][1] + 1)):
|
||||
with self.subTest(length=len(changed)):
|
||||
self.image_path.write_bytes(changed)
|
||||
self.argument('--image-sha256', fixture.digest(changed))
|
||||
self.refuse(before_prepare=True)
|
||||
|
||||
def test_image_chip_capacity_dio_and_descriptor_refused_before_prepare(self):
|
||||
images = [image(self.capacity, chip=5), image(0x1000000), image(self.capacity, mode=0)]
|
||||
bad_descriptor = bytearray(self.image)
|
||||
bad_descriptor[32] ^= 1
|
||||
images.append(bytes(bad_descriptor))
|
||||
for changed in images:
|
||||
with self.subTest(header=changed[:4].hex()):
|
||||
self.image_path.write_bytes(changed)
|
||||
self.argument('--image-sha256', fixture.digest(changed))
|
||||
self.refuse(before_prepare=True)
|
||||
|
||||
def test_app_readback_fault_or_changed_table_ota_prevents_runtime_finish(self):
|
||||
for env in ({'MOCK_BAD_READBACK': '1'}, {'MOCK_MUTATE_AFTER_WRITE': '0x8000'},
|
||||
{'MOCK_MUTATE_AFTER_WRITE': hex(self.ota_address)}):
|
||||
with self.subTest(fault=env):
|
||||
self.configure()
|
||||
result = self.run_tool(**env)
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertEqual(sum('write-flash' in line for line in self.actions), 1)
|
||||
self.assertNotIn('finish', self.actions)
|
||||
|
||||
def test_nonzero_infrastructure_target_keeps_existing_interface(self):
|
||||
self.image = image(self.capacity, target=1234)
|
||||
self.image_path.write_bytes(self.image)
|
||||
self.argument('--image-sha256', fixture.digest(self.image))
|
||||
for option in ('--expect-target-id', '--expect-hardware-id'):
|
||||
index = self.args.index(option)
|
||||
del self.args[index:index + 2]
|
||||
self.assert_success()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,44 @@
|
||||
"""Offline backup ownership guard; never inspect or open a real serial port."""
|
||||
from pathlib import Path
|
||||
import re
|
||||
import subprocess
|
||||
import unittest
|
||||
|
||||
TOOL = Path(__file__).resolve().parents[1] / "tools/hil/mcfirmware_bench_backup.sh"
|
||||
|
||||
|
||||
class BackupOwnershipTests(unittest.TestCase):
|
||||
def guard(self, status, stdout="", stderr=""):
|
||||
source = TOOL.read_text()
|
||||
guard = re.search(r"^backup_assert_unowned\(\) \{\n.*?^\}", source, re.M | re.S).group()
|
||||
# sudo/fuser are inert functions: an unprivileged fallback fails this
|
||||
# contract, including when its quiet return code resembles no owner.
|
||||
script = r'''
|
||||
set -eu
|
||||
backup_die() { printf '%s\n' "$*" >&2; exit 9; }
|
||||
fuser() { printf 'unguarded_fuser_called\n' >&2; return 1; }
|
||||
sudo() {
|
||||
[ "$#" = 3 ] && [ "$1" = -n ] && [ "$2" = fuser ] && [ "$3" = /dev/ttyACM999 ] || exit 8
|
||||
printf '%s' "$FAKE_STDOUT"
|
||||
printf '%s' "$FAKE_STDERR" >&2
|
||||
return "$FAKE_STATUS"
|
||||
}
|
||||
'''
|
||||
return subprocess.run(["bash", "-c", script + guard + "\nbackup_assert_unowned /dev/ttyACM999\n"],
|
||||
env={"FAKE_STATUS": str(status), "FAKE_STDOUT": stdout,
|
||||
"FAKE_STDERR": stderr}, capture_output=True, text=True)
|
||||
|
||||
def test_root_capable_quiet_no_owner_probe_is_accepted(self):
|
||||
result = self.guard(1)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
|
||||
def test_owner_permission_failure_or_ambiguous_probe_refuses(self):
|
||||
for status, stdout, stderr in ((0, "12345", ""), (1, "", "sudo: a password is required"),
|
||||
(1, "", "Usage: fuser"), (1, "12345", ""), (2, "", "")):
|
||||
with self.subTest(status=status, has_stdout=bool(stdout), has_stderr=bool(stderr)):
|
||||
result = self.guard(status, stdout, stderr)
|
||||
self.assertEqual(result.returncode, 9)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,274 @@
|
||||
"""Offline guarded-flasher contract tests: no device or PlatformIO process.
|
||||
|
||||
The reviewed real flasher and parser are pinned runtime inputs, outside this
|
||||
repository. This fixture supplies a separately pinned, pure parser and inert
|
||||
flasher functions; serial commands operate only a temporary bytearray file.
|
||||
"""
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import struct
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
TOOL = ROOT / "tools/hil/mcfirmware_bench_data_restore.sh"
|
||||
MAC = "441bf669c9c0"
|
||||
CAPACITY = 0x800000
|
||||
NVS = (0x9000, 0x5000)
|
||||
SPIFFS = (0x30000, 0x1000)
|
||||
APP = (0x10000, 256)
|
||||
|
||||
PARSER = r'''
|
||||
import hashlib,json,re,struct,zlib
|
||||
class ExpansionError(Exception): pass
|
||||
def safe_json(data): return json.loads(data)
|
||||
def parse_table(data,capacity):
|
||||
if len(data)!=4096: raise ExpansionError('bad table length')
|
||||
result=[]
|
||||
for offset in range(0,len(data),32):
|
||||
row=data[offset:offset+32]
|
||||
if row[:2]==b'\xff\xff': break
|
||||
if row[:2]!=b'\xaa\x50': raise ExpansionError('bad table entry')
|
||||
kind,subtype,address,size,name,flags=struct.unpack_from('<BBII16sI',row,2)
|
||||
if address<0x9000 or address%4096 or size%4096 or not size or address+size>capacity:
|
||||
raise ExpansionError('bad table geometry')
|
||||
result.append(dict(type=kind,subtype=subtype,offset=address,size=size,name=name.rstrip(b'\0').decode(),flags=flags))
|
||||
ordered=sorted(result,key=lambda p:p['offset'])
|
||||
if any(a['offset']+a['size']>b['offset'] for a,b in zip(ordered,ordered[1:])):
|
||||
raise ExpansionError('overlapping table')
|
||||
return result
|
||||
def active_ota_offset(parts,data):
|
||||
if len(data)!=8192: raise ExpansionError('bad OTA size')
|
||||
apps=sorted((p for p in parts if p['type']==0),key=lambda p:p['subtype'])
|
||||
if data==b'\xff'*8192: return apps[0]['offset']
|
||||
sequence=struct.unpack_from('<I',data)[0]
|
||||
crc=struct.unpack_from('<I',data,28)[0]
|
||||
if crc!=(zlib.crc32(struct.pack('<I',sequence),0xffffffff)&0xffffffff):
|
||||
raise ExpansionError('bad OTA CRC')
|
||||
return apps[(sequence-1)%len(apps)]['offset']
|
||||
def flash_identity(text):
|
||||
chip=9 if 'ESP32-S3' in text else 5
|
||||
return int(re.search(r'flash size:(\d+)MB',text)[1])*1048576,chip
|
||||
'''
|
||||
|
||||
FLASHER = r'''
|
||||
esptool_set_variables() {
|
||||
NORESET=no-reset;HARDRESET=hard-reset;READMAC=read-mac;READFLASH=read-flash;WRITEFLASH=write-flash;FLASHID=flash-id
|
||||
}
|
||||
save_selected_serial_port() {
|
||||
DEVICE_PORT=$1
|
||||
}
|
||||
prepare_esp32_flash_session() {
|
||||
printf 'prepare\n' >> "$MOCK_ACTIONS"
|
||||
ESP32_FLASH_EXPECTED_MAC=441bf669c9c0;ESP32_SESSION_IS_S3=1
|
||||
}
|
||||
selected_flash_serial_port() {
|
||||
printf '%s\n' "$1"
|
||||
}
|
||||
esp32_record_and_verify_probe_output() {
|
||||
[[ "$ESP32_FLASH_EXPECTED_MAC" == 441bf669c9c0 ]]
|
||||
}
|
||||
esp32_verified_destructive_port() {
|
||||
printf '%s\n' "$1"
|
||||
}
|
||||
esp32_write_after_mode() {
|
||||
printf '%s\n' "${MOCK_AFTER:-no-reset}"
|
||||
}
|
||||
finish_esp32_flash_session() {
|
||||
printf 'finish\n' >> "$MOCK_ACTIONS"
|
||||
}
|
||||
run_esp32_session_esptool() {
|
||||
printf '%s\n' "$*" >> "$MOCK_ACTIONS"
|
||||
shift
|
||||
[[ "$1" == --after && ( "$2" == no-reset || "$2" == hard-reset ) && "$3" == --baud && "$4" == 115200 ]] || return 97
|
||||
shift 4
|
||||
case "$1" in
|
||||
flash-id) printf 'Chip is %s\nMAC:44:1b:f6:69:c9:c0\nDetected flash size:8MB\n' "${MOCK_CHIP:-ESP32-S3}" ;;
|
||||
read-mac) printf 'MAC:44:1b:f6:69:c9:c0\n' ;;
|
||||
read-flash) python3 - "$MOCK_FLASH" "$2" "$3" "$4" <<'READ'
|
||||
import os,sys
|
||||
with open(sys.argv[1],'rb') as src:
|
||||
src.seek(int(sys.argv[2],0)); data=src.read(int(sys.argv[3],0))
|
||||
if os.environ.get('MOCK_BAD_READBACK') and 'nvs-readback' in sys.argv[4]:
|
||||
data=bytes([data[0]^1])+data[1:]
|
||||
open(sys.argv[4],'wb').write(data)
|
||||
READ
|
||||
;;
|
||||
write-flash) python3 - "$MOCK_FLASH" "$2" "$3" <<'WRITE'
|
||||
import os,sys
|
||||
with open(sys.argv[1],'r+b') as dst:
|
||||
dst.seek(int(sys.argv[2],0));dst.write(open(sys.argv[3],'rb').read())
|
||||
if int(sys.argv[2],0)==0x9000 and os.environ.get('MOCK_MUTATE_AFTER_WRITE'):
|
||||
at=int(os.environ['MOCK_MUTATE_AFTER_WRITE'],0)
|
||||
dst.seek(at);old=dst.read(1);dst.seek(at);dst.write(bytes([old[0]^1]))
|
||||
WRITE
|
||||
;;
|
||||
*) return 98 ;;
|
||||
esac
|
||||
}
|
||||
'''
|
||||
|
||||
|
||||
def digest(data):
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
@unittest.skipUnless(os.name == "posix" and all(shutil.which(c) for c in
|
||||
("bash", "python3", "awk", "stat", "sha256sum", "realpath", "timeout")),
|
||||
"POSIX shell tools required; no physical device required")
|
||||
class DataRestoreTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory(prefix="mcfirmware-data-offline-")
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.root = Path(self.temp.name)
|
||||
self.root.chmod(0o700)
|
||||
self.bin = self.root / "bin"
|
||||
self.bin.mkdir()
|
||||
scripts = {"readlink": "#!/bin/sh\nprintf '%s\\n' /dev/null\n",
|
||||
"fuser": "#!/bin/sh\nexit 1\n", "pipx": "#!/bin/sh\nexit 99\n",
|
||||
"udevadm": "#!/bin/sh\nexit 99\n",
|
||||
"sudo": '#!/bin/sh\n[ "$1" = -n ] && [ "$2" = fuser ] || exit 98\nshift\nexec "$@"\n'}
|
||||
for name, text in scripts.items():
|
||||
p = self.bin / name
|
||||
p.write_text(text)
|
||||
p.chmod(0o700)
|
||||
self.flasher = self.root / "flasher/mcfirmware.sh"
|
||||
helper = self.flasher.parent / "tools/meshcore_partition_expansion.py"
|
||||
helper.parent.mkdir(parents=True)
|
||||
helper.write_text(PARSER)
|
||||
self.flasher.write_text('MESHCORE_PARTITION_EXPANSION_TOOL_SHA256="' + digest(helper.read_bytes()) + '"\n'
|
||||
+ 'MESHCORE_USB_RESET_TOOL_SHA256="' + '0' * 64 + '"\n'
|
||||
+ 'MESHCORE_ESP32_BOOTLOADER_TOOL_SHA256="' + '0' * 64 + '"\n' + FLASHER
|
||||
+ ''.join(f'inert_{i}() {{\n :\n}}\n' for i in range(100)))
|
||||
table = b"".join(struct.pack('<2sBBII16sI', b'\xaa\x50', kind, subtype, offset, length, name.encode(), 0)
|
||||
for name, kind, subtype, offset, length in [
|
||||
('nvs', 1, 2, *NVS), ('otadata', 1, 0, 0xe000, 0x2000),
|
||||
('app0', 0, 0x10, 0x10000, 0x10000), ('app1', 0, 0x11, 0x20000, 0x10000),
|
||||
('spiffs', 1, 0x82, *SPIFFS)])
|
||||
self.table = table.ljust(4096, b'\xff')
|
||||
backup = bytearray(b'\xff' * CAPACITY)
|
||||
backup[0x8000:0x9000] = self.table
|
||||
backup[NVS[0]:sum(NVS)] = b'\x11' * NVS[1]
|
||||
backup[SPIFFS[0]:sum(SPIFFS)] = b'\x22' * SPIFFS[1]
|
||||
self.backup = self.root / "backup.private.bin"
|
||||
self.backup.write_bytes(backup)
|
||||
self.backup.chmod(0o600)
|
||||
self.manifest = self.root / "backup.private.bin.json"
|
||||
self.manifest.write_text(json.dumps(dict(complete=True, operation='read-only-flash-backup', chip='ESP32-S3',
|
||||
device_mac=MAC, offset=0, length=CAPACITY, flash_bytes=CAPACITY, sha256=digest(backup), artifact=self.backup.name)))
|
||||
self.manifest.chmod(0o600)
|
||||
self.app = bytes(range(256))
|
||||
live = bytearray(backup)
|
||||
live[NVS[0]:sum(NVS)] = b'\x33' * NVS[1]
|
||||
live[SPIFFS[0]:sum(SPIFFS)] = b'\x44' * SPIFFS[1]
|
||||
live[APP[0]:sum(APP)] = self.app
|
||||
self.before = bytes(live)
|
||||
self.flash = self.root / "live-flash.bin"
|
||||
self.flash.write_bytes(live)
|
||||
self.trace = self.root / "actions"
|
||||
self.bench = self.root / "bench"
|
||||
self.bench.mkdir(mode=0o700)
|
||||
self.args = ['bash', str(TOOL), '--port', '/dev/ttyACM99', '--usb-path', 'exact-path', '--usb-id', 'exact-id',
|
||||
'--device', 'MOCK', '--bench-dir', str(self.bench), '--expect-mac', MAC,
|
||||
'--backup', str(self.backup), '--backup-manifest', str(self.manifest), '--backup-sha256', digest(backup),
|
||||
'--expect-partition-sha256', digest(self.table), '--app-address', hex(APP[0]),
|
||||
'--app-length', str(APP[1]), '--expect-app-sha256', digest(self.app), '--allow-write',
|
||||
'--flasher', str(self.flasher), '--flasher-sha256', digest(self.flasher.read_bytes())]
|
||||
|
||||
def run_tool(self, **env):
|
||||
result = subprocess.run(self.args, env=dict(os.environ, PATH=str(self.bin) + os.pathsep + os.environ['PATH'],
|
||||
MOCK_FLASH=str(self.flash), MOCK_ACTIONS=str(self.trace), **env), capture_output=True, text=True, timeout=15)
|
||||
self.actions = self.trace.read_text().splitlines() if self.trace.exists() else []
|
||||
return result
|
||||
|
||||
def assert_no_writes(self, result):
|
||||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||||
self.assertFalse(any('write-flash' in line for line in self.actions), self.actions)
|
||||
self.assertNotIn('finish', self.actions)
|
||||
|
||||
def test_only_original_nvs_and_spiffs_written_and_every_other_byte_preserved(self):
|
||||
result = self.run_tool()
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
writes = [line.split() for line in self.actions if 'write-flash' in line]
|
||||
self.assertEqual([int(line[6]) for line in writes], [NVS[0], SPIFFS[0]])
|
||||
after = self.flash.read_bytes()
|
||||
expected = bytearray(self.before)
|
||||
original = self.backup.read_bytes()
|
||||
for offset, length in (NVS, SPIFFS):
|
||||
expected[offset:offset + length] = original[offset:offset + length]
|
||||
self.assertEqual(after, expected)
|
||||
self.assertEqual(self.actions[-1], 'finish')
|
||||
state = next(self.bench.glob('.mcfirmware-data-restore.*'))
|
||||
report = json.loads((state / 'result.json').read_text())
|
||||
self.assertTrue(report['active_application_unchanged'] and report['data_readbacks_verified'])
|
||||
self.assertEqual({p.stat().st_mode & 0o777 for p in state.iterdir() if p.is_file()}, {0o600})
|
||||
|
||||
def test_uart_final_probe_reset_then_real_finish(self):
|
||||
result = self.run_tool(MOCK_AFTER='hard-reset')
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertIn('--after hard-reset --baud 115200 read-mac', self.actions[-2])
|
||||
self.assertEqual(self.actions[-1], 'finish')
|
||||
|
||||
def test_changed_live_table_refused_before_data_writes(self):
|
||||
data = bytearray(self.flash.read_bytes())
|
||||
data[0x800c] ^= 1
|
||||
self.flash.write_bytes(data)
|
||||
self.assert_no_writes(self.run_tool())
|
||||
|
||||
def test_changed_live_ota_refused_before_data_writes(self):
|
||||
data = bytearray(self.flash.read_bytes())
|
||||
data[0xe000] = 1
|
||||
self.flash.write_bytes(data)
|
||||
self.assert_no_writes(self.run_tool())
|
||||
|
||||
def test_changed_live_current_app_refused_before_data_writes(self):
|
||||
data = bytearray(self.flash.read_bytes())
|
||||
data[APP[0]] ^= 1
|
||||
self.flash.write_bytes(data)
|
||||
self.assert_no_writes(self.run_tool())
|
||||
|
||||
def test_wrong_chip_refused_before_data_writes(self):
|
||||
self.assert_no_writes(self.run_tool(MOCK_CHIP='ESP32-C3'))
|
||||
|
||||
def test_wrong_backup_hash_refused_before_prepare(self):
|
||||
self.args[self.args.index('--backup-sha256') + 1] = '0' * 64
|
||||
self.assert_no_writes(self.run_tool())
|
||||
self.assertNotIn('prepare', self.actions)
|
||||
|
||||
def test_wrong_mac_refused_before_data_writes(self):
|
||||
self.args[self.args.index('--expect-mac') + 1] = '112233445566'
|
||||
self.assert_no_writes(self.run_tool())
|
||||
|
||||
def test_first_readback_fault_prevents_second_write_and_runtime_finish(self):
|
||||
result = self.run_tool(MOCK_BAD_READBACK='1')
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
writes = [line.split() for line in self.actions if 'write-flash' in line]
|
||||
self.assertEqual([int(line[6]) for line in writes], [NVS[0]])
|
||||
self.assertNotIn('finish', self.actions)
|
||||
|
||||
def test_postwrite_table_ota_app_mutations_prevent_runtime_finish(self):
|
||||
for address in (0x8000, 0xe000, APP[0]):
|
||||
with self.subTest(changed_address=address):
|
||||
self.flash.write_bytes(self.before)
|
||||
self.trace.unlink(missing_ok=True)
|
||||
result = self.run_tool(MOCK_MUTATE_AFTER_WRITE=hex(address))
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertNotIn('finish', self.actions)
|
||||
|
||||
def test_wrong_active_app_address_refused_before_prepare(self):
|
||||
self.args[self.args.index('--app-address') + 1] = '0x20000'
|
||||
self.assert_no_writes(self.run_tool())
|
||||
self.assertNotIn('prepare', self.actions)
|
||||
|
||||
def test_root_owner_or_sudo_error_refused_before_prepare(self):
|
||||
(self.bin / 'sudo').write_text('#!/bin/sh\nprintf "owner inspection unavailable" >&2\nexit 1\n')
|
||||
self.assert_no_writes(self.run_tool())
|
||||
self.assertNotIn('prepare', self.actions)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,302 @@
|
||||
"""Offline owned-CDP startup regressions; never open a browser or serial port."""
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from types import SimpleNamespace
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
HIL = Path(__file__).resolve().parents[1] / "tools/hil"
|
||||
sys.path.insert(0, str(HIL))
|
||||
spec = importlib.util.spec_from_file_location("official_acceptance", HIL / "official_console_acceptance.py")
|
||||
acceptance = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(acceptance)
|
||||
|
||||
|
||||
class OwnedStartupTests(unittest.TestCase):
|
||||
def fixture(self, *, fail_first_page=False, wrong_profile=False, pages=1):
|
||||
requests, sockets, connections = [], [], []
|
||||
fail = [fail_first_page]
|
||||
profile = Path("/private/owned-profile")
|
||||
arguments = ["--user-data-dir=" + str(profile), "--remote-debugging-port=12345"]
|
||||
if wrong_profile:
|
||||
arguments[0] = "--user-data-dir=/private/another-browser"
|
||||
|
||||
class Socket:
|
||||
def __init__(self, browser):
|
||||
self.browser, self.closed, self.queue = browser, False, []
|
||||
sockets.append(self)
|
||||
|
||||
def send(self, text):
|
||||
request = json.loads(text)
|
||||
if self.browser:
|
||||
# Browser events must not be mistaken for the command reply.
|
||||
self.queue.extend([{"method": "Target.targetCreated"},
|
||||
{"id": request["id"], "result": {"arguments": arguments}}])
|
||||
else:
|
||||
if fail[0]:
|
||||
fail[0] = False
|
||||
raise OSError("renderer connection interrupted")
|
||||
self.queue.append({"id": request["id"], "result": {}})
|
||||
|
||||
def recv(self):
|
||||
return json.dumps(self.queue.pop(0))
|
||||
|
||||
def settimeout(self, timeout):
|
||||
self.timeout = timeout
|
||||
|
||||
def close(self):
|
||||
self.closed = True
|
||||
|
||||
def connect(url, **kwargs):
|
||||
connections.append((url, kwargs))
|
||||
return Socket("/browser/" in url)
|
||||
|
||||
class Opener:
|
||||
def open(self, request, timeout):
|
||||
requests.append(request)
|
||||
if request.endswith("/json/version"):
|
||||
value = {"webSocketDebuggerUrl": "ws://127.0.0.1:12345/browser/owned", "Browser": "test Chromium"}
|
||||
elif request.endswith("/json/list"):
|
||||
value = [{"type": "page", "url": "about:blank",
|
||||
"webSocketDebuggerUrl": f"ws://127.0.0.1:12345/page/{i}"} for i in range(pages)]
|
||||
else:
|
||||
raise AssertionError("Startup must reuse its initial page, never PUT /json/new")
|
||||
return io.StringIO(json.dumps(value))
|
||||
|
||||
websocket = SimpleNamespace(create_connection=connect,
|
||||
WebSocketTimeoutException=type("SocketTimeout", (Exception,), {}),
|
||||
WebSocketConnectionClosedException=type("SocketClosed", (Exception,), {}))
|
||||
return profile, websocket, Opener(), requests, sockets, connections
|
||||
|
||||
def test_reuses_initial_page_and_matches_owned_reply_after_event(self):
|
||||
profile, websocket, opener, requests, sockets, connections = self.fixture()
|
||||
with patch.object(acceptance.urllib.request, "build_opener", return_value=opener):
|
||||
page = acceptance.CDP(websocket, 12345, profile)
|
||||
self.assertEqual(requests, ["http://127.0.0.1:12345/json/version", "http://127.0.0.1:12345/json/list"])
|
||||
self.assertEqual(page.browser_version, "test Chromium")
|
||||
self.assertTrue(sockets[0].closed)
|
||||
self.assertFalse(sockets[1].closed)
|
||||
for _, kwargs in connections:
|
||||
self.assertEqual(kwargs["http_no_proxy"], ["127.0.0.1", "localhost"])
|
||||
page.socket.close()
|
||||
|
||||
def test_partial_page_socket_is_closed_and_retry_creates_no_new_target(self):
|
||||
profile, websocket, opener, requests, sockets, _ = self.fixture(fail_first_page=True)
|
||||
browser = SimpleNamespace(poll=lambda: None)
|
||||
report = {}
|
||||
with patch.object(acceptance.urllib.request, "build_opener", return_value=opener), \
|
||||
patch.object(acceptance.time, "sleep"):
|
||||
page = acceptance.wait_for_owned_browser(websocket, 12345, profile, browser, report)
|
||||
self.assertEqual(report["browser_startup"]["attempts"], 2)
|
||||
self.assertEqual(report["browser_startup"]["last_phase"], "initial_page_enable")
|
||||
self.assertTrue(sockets[0].closed and sockets[1].closed and sockets[2].closed)
|
||||
self.assertFalse(sockets[3].closed)
|
||||
self.assertEqual(sum(request.endswith("/json/list") for request in requests), 2)
|
||||
self.assertTrue(all(isinstance(request, str) for request in requests))
|
||||
page.socket.close()
|
||||
|
||||
def test_another_owned_profile_is_rejected_before_page_attach(self):
|
||||
profile, websocket, opener, requests, sockets, _ = self.fixture(wrong_profile=True)
|
||||
with patch.object(acceptance.urllib.request, "build_opener", return_value=opener), \
|
||||
self.assertRaisesRegex(RuntimeError, "does_not_match_owned_profile"):
|
||||
acceptance.CDP(websocket, 12345, profile)
|
||||
self.assertEqual(len(requests), 1)
|
||||
self.assertTrue(sockets[0].closed)
|
||||
|
||||
def test_extra_pages_are_rejected_without_adding_another_page(self):
|
||||
profile, websocket, opener, _, sockets, _ = self.fixture(pages=2)
|
||||
with patch.object(acceptance.urllib.request, "build_opener", return_value=opener), \
|
||||
self.assertRaisesRegex(RuntimeError, "one_initial_blank_page"):
|
||||
acceptance.CDP(websocket, 12345, profile)
|
||||
self.assertEqual(len(sockets), 1)
|
||||
self.assertTrue(sockets[0].closed)
|
||||
|
||||
def test_startup_deadline_preserves_phase_without_arbitrary_exception_text(self):
|
||||
_, websocket, _, _, _, _ = self.fixture()
|
||||
error = OSError("private host details")
|
||||
error.startup_phase = "version_endpoint"
|
||||
report = {}
|
||||
with patch.object(acceptance, "CDP", side_effect=error), \
|
||||
patch.object(acceptance.time, "monotonic", side_effect=[0, 0, .001, .02]), \
|
||||
patch.object(acceptance.time, "sleep"), \
|
||||
self.assertRaisesRegex(RuntimeError, "owned_Chromium_startup_deadline"):
|
||||
acceptance.wait_for_owned_browser(websocket, 12345, Path("/owned"),
|
||||
SimpleNamespace(poll=lambda: None), report, timeout=.01)
|
||||
self.assertEqual(report["browser_startup"]["attempts"], 2)
|
||||
self.assertEqual(report["browser_startup"]["last_phase"], "version_endpoint")
|
||||
self.assertNotIn("private host details", json.dumps(report))
|
||||
|
||||
|
||||
class OwnedCleanupTests(unittest.TestCase):
|
||||
def test_termination_unwinds_and_restores_original_handler(self):
|
||||
previous = signal.getsignal(signal.SIGTERM)
|
||||
unwound = []
|
||||
with self.assertRaises(acceptance.NativeConsoleTerminated):
|
||||
with acceptance.cleanup_on_sigterm():
|
||||
try:
|
||||
signal.getsignal(signal.SIGTERM)(signal.SIGTERM, None)
|
||||
finally:
|
||||
unwound.append(True)
|
||||
self.assertEqual(signal.getsignal(signal.SIGTERM), signal.SIG_IGN)
|
||||
self.assertEqual(unwound, [True])
|
||||
self.assertIs(signal.getsignal(signal.SIGTERM), previous)
|
||||
|
||||
def test_normal_exit_restores_original_handler(self):
|
||||
previous = signal.getsignal(signal.SIGTERM)
|
||||
with acceptance.cleanup_on_sigterm():
|
||||
self.assertTrue(callable(signal.getsignal(signal.SIGTERM)))
|
||||
self.assertIs(signal.getsignal(signal.SIGTERM), previous)
|
||||
|
||||
def test_socket_and_stream_failure_still_closes_only_owned_browser(self):
|
||||
from unittest.mock import Mock
|
||||
page = SimpleNamespace(evaluate=Mock(side_effect=RuntimeError("private stream details")),
|
||||
socket=SimpleNamespace(close=Mock(side_effect=OSError("private socket details"))))
|
||||
browser = SimpleNamespace(pid=123456, poll=Mock(side_effect=[None, 0]), wait=Mock())
|
||||
report = {}
|
||||
with patch.object(acceptance.os, "killpg") as kill:
|
||||
acceptance.close_owned_browser(page, browser, report)
|
||||
kill.assert_called_once_with(123456, signal.SIGTERM)
|
||||
browser.wait.assert_called_once_with(timeout=10)
|
||||
self.assertEqual(report, {"browser_stream_cleanup_error": True,
|
||||
"browser_socket_cleanup_error": True, "owned_browser_closed": True})
|
||||
self.assertNotIn("private", json.dumps(report))
|
||||
|
||||
def test_interrupted_stream_cleanup_still_closes_owned_browser(self):
|
||||
from unittest.mock import Mock
|
||||
page = SimpleNamespace(evaluate=Mock(side_effect=acceptance.NativeConsoleTerminated()),
|
||||
socket=SimpleNamespace(close=Mock()))
|
||||
browser = SimpleNamespace(pid=123456, poll=Mock(side_effect=[None, 0]), wait=Mock())
|
||||
with patch.object(acceptance.os, "killpg") as kill, \
|
||||
self.assertRaises(acceptance.NativeConsoleTerminated):
|
||||
acceptance.close_owned_browser(page, browser, {})
|
||||
page.socket.close.assert_called_once_with()
|
||||
kill.assert_called_once_with(123456, signal.SIGTERM)
|
||||
|
||||
def test_unresponsive_owned_browser_uses_bounded_kill_fallback(self):
|
||||
from unittest.mock import Mock, call
|
||||
browser = SimpleNamespace(pid=123456, poll=Mock(side_effect=[None, -9]),
|
||||
wait=Mock(side_effect=[subprocess.TimeoutExpired("owned", 10), None]))
|
||||
report = {}
|
||||
with patch.object(acceptance.os, "killpg") as kill:
|
||||
acceptance.close_owned_browser(None, browser, report)
|
||||
self.assertEqual(kill.call_args_list, [call(123456, signal.SIGTERM), call(123456, signal.SIGKILL)])
|
||||
self.assertEqual(browser.wait.call_args_list, [call(timeout=10), call(timeout=5)])
|
||||
self.assertTrue(report["owned_browser_closed"])
|
||||
|
||||
def test_termination_produces_private_failure_report_and_143(self):
|
||||
def interrupted(_args, report):
|
||||
try:
|
||||
signal.getsignal(signal.SIGTERM)(signal.SIGTERM, None)
|
||||
finally:
|
||||
report["mock_owned_resources_unwound"] = True
|
||||
|
||||
previous = signal.getsignal(signal.SIGTERM)
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
work = Path(directory)
|
||||
work.chmod(0o700)
|
||||
output = work / "result.json"
|
||||
with patch.object(acceptance, "execute", side_effect=interrupted), \
|
||||
patch("sys.stdout", new=io.StringIO()):
|
||||
status = acceptance.main(["--serial", "exact", "--expected-board", "board",
|
||||
"--expected-version", "version", "--logging", "off", "--console-source", "/unused",
|
||||
"--artifact-manifest", "/unused", "--output", str(output),
|
||||
"--work-dir", str(work), "--execute"])
|
||||
report = json.loads(output.read_text())
|
||||
self.assertEqual(status, 143)
|
||||
self.assertFalse(report["ok"])
|
||||
self.assertEqual(report["termination_signal"], "SIGTERM")
|
||||
self.assertTrue(report["mock_owned_resources_unwound"])
|
||||
self.assertEqual(output.stat().st_mode & 0o777, 0o600)
|
||||
self.assertIs(signal.getsignal(signal.SIGTERM), previous)
|
||||
|
||||
|
||||
class SafeJavaScriptFailureTests(unittest.TestCase):
|
||||
def evaluate_error(self, details):
|
||||
page = object.__new__(acceptance.CDP)
|
||||
page.call = lambda *args, **kwargs: {"exceptionDetails": details}
|
||||
with self.assertRaises(acceptance.NativeConsoleJavaScriptError) as caught:
|
||||
page.evaluate("hil.query('board')")
|
||||
return caught.exception
|
||||
|
||||
def test_page_literal_error_is_reported_without_stack_or_description(self):
|
||||
error = self.evaluate_error({"text": "Uncaught (in promise)", "exception": {
|
||||
"className": "Error", "description": "Error: framed_reply_deadline\n at private-site/secret"}})
|
||||
self.assertEqual(error.safe_details, {"class": "Error", "code": "framed_reply_deadline"})
|
||||
self.assertEqual(str(error), "native_console_javascript_failed:framed_reply_deadline")
|
||||
self.assertNotIn("secret", json.dumps(error.safe_details))
|
||||
|
||||
def test_dom_exception_name_is_safe_but_native_message_is_hashed(self):
|
||||
message = "NetworkError: private serial endpoint secret"
|
||||
error = self.evaluate_error({"exception": {"className": "DOMException", "description": message}})
|
||||
self.assertEqual(error.safe_details["class"], "DOMException")
|
||||
self.assertEqual(error.safe_details["dom_exception_name"], "NetworkError")
|
||||
self.assertEqual(error.safe_details["message_sha256"], acceptance.hashlib.sha256(message.encode()).hexdigest())
|
||||
self.assertNotIn("secret", str(error) + json.dumps(error.safe_details))
|
||||
|
||||
def test_unknown_class_and_known_code_prefix_do_not_expose_arbitrary_text(self):
|
||||
message = "Error: framed_reply_deadline secret-password"
|
||||
error = self.evaluate_error({"exception": {"className": "private-class-secret", "description": message}})
|
||||
self.assertEqual(error.safe_details["class"], "unrecognized")
|
||||
self.assertEqual(error.safe_details["code"], "unclassified_exception")
|
||||
self.assertEqual(error.safe_details["message_sha256"], acceptance.hashlib.sha256(message.encode()).hexdigest())
|
||||
self.assertNotIn("secret", str(error) + json.dumps(error.safe_details))
|
||||
|
||||
@unittest.skipUnless(shutil.which("node"), "Node required for offline PAGE protocol test")
|
||||
def test_page_accepts_exact_prompt_echo_and_requires_complete_response(self):
|
||||
# Execute the actual PAGE module body with an inert SDK-shaped class.
|
||||
# navigator.serial returns a plain object; no browser/device is opened.
|
||||
script = acceptance.PAGE.split('<script type="module">', 1)[1].split('</script>', 1)[0]
|
||||
self.assertIn("import {SerialConsole} from './console.js';", script)
|
||||
script = script.replace("import {SerialConsole} from './console.js';", "")
|
||||
setup = r'''
|
||||
import assert from 'node:assert/strict';
|
||||
globalThis.window=globalThis;
|
||||
Object.defineProperty(globalThis,'navigator',{value:{serial:{getPorts:async()=>[
|
||||
{getInfo:()=>({usbVendorId:0x303a,usbProductId:0x1001})}
|
||||
]}}});
|
||||
const realTimeout=globalThis.setTimeout;
|
||||
// Only test deadlines are shortened; idle remains a no-transmit timer.
|
||||
globalThis.setTimeout=(fn,ms)=>realTimeout(fn,ms===10000?5:ms);
|
||||
let echoPrefix='> ', complete=true;
|
||||
class SerialConsole {
|
||||
constructor(port){this.port=port;this.connected=false;}
|
||||
async connect(){this.connected=true;}
|
||||
async disconnect(){this.connected=false;}
|
||||
async reset(){throw Error('reset_must_never_be_called');}
|
||||
async sendCommand(command){
|
||||
this.onOutput(echoPrefix+command+'\n');
|
||||
this.onOutput(' -> bench reply'+(complete?'\n':''));
|
||||
}
|
||||
}
|
||||
'''
|
||||
checks = r'''
|
||||
await hil.open(0x303a,0x1001);
|
||||
for(const prefix of ['> ','>','']){
|
||||
echoPrefix=prefix;
|
||||
const result=await hil.query('board');
|
||||
assert.equal(result.body,'bench reply');
|
||||
assert.equal(result.complete_CRLF_reply,true);
|
||||
}
|
||||
const writes=hil.writes;
|
||||
await hil.idle(0);
|
||||
assert.equal(hil.writes,writes);
|
||||
echoPrefix='> stale ';
|
||||
await assert.rejects(hil.query('board'),/framed_reply_deadline/);
|
||||
echoPrefix='> ';complete=false;
|
||||
await assert.rejects(hil.query('board'),/framed_reply_deadline/);
|
||||
await hil.close();
|
||||
'''
|
||||
ran = subprocess.run([shutil.which("node"), "--input-type=module", "-"],
|
||||
input=setup + script + checks, capture_output=True, text=True, timeout=10)
|
||||
self.assertEqual(ran.returncode, 0, ran.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user