222 lines
12 KiB
Python
222 lines
12 KiB
Python
"""Disposable UF2/USB fixtures. Never opens a host serial/block device."""
|
|
import errno
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
import struct
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
import radio_uf2 as uf2
|
|
import c01_radio_owner as owner
|
|
|
|
|
|
def firmware(blocks=2):
|
|
result = bytearray()
|
|
for number in range(blocks):
|
|
header = struct.pack('<8I', 0x0A324655, 0x9E5D5157, 0x2000,
|
|
0x26000 + number * 256, 256, number, blocks, 0xADA52840)
|
|
result += header + bytes([number % 256]) * 256 + bytes(220) + struct.pack('<I', 0x0AB16F30)
|
|
return bytes(result)
|
|
|
|
|
|
class FirmwareTests(unittest.TestCase):
|
|
def test_application_only_image(self):
|
|
raw = firmware()
|
|
value = uf2.validate_image(raw)
|
|
self.assertEqual(value['sha256'], hashlib.sha256(raw).hexdigest())
|
|
self.assertEqual(value['blocks'], 2)
|
|
# Block ordering is not an authority; complete distinct regions are.
|
|
self.assertEqual(uf2.validate_image(raw[512:] + raw[:512])['blocks'], 2)
|
|
|
|
def test_corruption_family_reserved_regions_and_missing_blocks(self):
|
|
for offset, value in [(0, 0), (4, 0), (8, 0), (8, 0x2001), (28, 0),
|
|
(12, 0), (12, 0xF4000), (12, 0x26001), (16, 512),
|
|
(20, 2), (24, 3), (508, 0), (512 + 20, 0),
|
|
(512 + 12, 0x26000), (512 + 12, 0x28000)]:
|
|
with self.subTest(offset=offset, value=value):
|
|
raw = bytearray(firmware())
|
|
struct.pack_into('<I', raw, offset, value)
|
|
with self.assertRaises(ValueError):
|
|
uf2.validate_image(bytes(raw))
|
|
for raw in (b'', firmware()[:-1], firmware()[512:], b'x' * (uf2.MAX_IMAGE + 512)):
|
|
with self.subTest(size=len(raw)), self.assertRaises(ValueError):
|
|
uf2.validate_image(raw)
|
|
|
|
def test_bootloader_identity_and_softdevice(self):
|
|
board = b'UF2 Bootloader 0.9.2\r\nBoard-ID: WisBlock-RAK4631-Board\r\n'
|
|
uf2.validate_board_info(board)
|
|
uf2.validate_board_info(board + b'SoftDevice: S140 6.1.1\n')
|
|
for bad in (board.replace(b'RAK4631', b'RAK11300'), board * 2,
|
|
board + b'SoftDevice: S140 7.3.0\n', board + b'\xff', b'x' * 4097):
|
|
with self.subTest(raw=bad[:60]), self.assertRaises((ValueError, UnicodeError)):
|
|
uf2.validate_board_info(bad)
|
|
|
|
def test_copy_short_writes_and_receipt_is_not_verification(self):
|
|
with tempfile.TemporaryFile() as source, tempfile.TemporaryFile() as target:
|
|
source.write(firmware()); source.flush()
|
|
original = os.write
|
|
with patch.object(uf2.os, 'write', side_effect=lambda fd, data: original(fd, data[:37])):
|
|
receipt = uf2.copy_image(source.fileno(), target.fileno())
|
|
target.seek(0)
|
|
self.assertEqual(target.read(), firmware())
|
|
self.assertEqual(receipt['state'], 'written-awaiting-device-verification')
|
|
self.assertNotIn('verified', receipt)
|
|
|
|
def test_failure_never_returns_a_receipt(self):
|
|
for failure in (lambda *_: 0, OSError(errno.ENODEV, 'USB disconnected')):
|
|
with tempfile.TemporaryFile() as source, tempfile.TemporaryFile() as target:
|
|
source.write(firmware()); source.flush()
|
|
with patch.object(uf2.os, 'write', side_effect=failure), self.assertRaises((ValueError, OSError)):
|
|
uf2.copy_image(source.fileno(), target.fileno())
|
|
with tempfile.TemporaryFile() as source, tempfile.TemporaryFile() as target:
|
|
source.write(firmware()); source.flush()
|
|
with patch.object(uf2.os, 'fsync', side_effect=OSError(errno.EIO, 'flush failed')), self.assertRaises(OSError):
|
|
uf2.copy_image(source.fileno(), target.fileno())
|
|
|
|
def test_bad_image_and_existing_destination_are_untouched(self):
|
|
with tempfile.TemporaryFile() as source, tempfile.TemporaryFile() as target:
|
|
source.write(b'invalid'); source.flush()
|
|
with patch.object(uf2.os, 'write') as write, self.assertRaises(ValueError):
|
|
uf2.copy_image(source.fileno(), target.fileno())
|
|
write.assert_not_called()
|
|
with tempfile.TemporaryFile() as source, tempfile.TemporaryFile() as target:
|
|
source.write(firmware()); source.flush()
|
|
target.write(b'operator data'); target.flush()
|
|
with self.assertRaises(ValueError):
|
|
uf2.copy_image(source.fileno(), target.fileno())
|
|
target.seek(0)
|
|
self.assertEqual(target.read(), b'operator data')
|
|
|
|
|
|
class OwnerTests(unittest.TestCase):
|
|
def setUp(self):
|
|
master, slave = os.openpty()
|
|
self.addCleanup(os.close, master); self.addCleanup(os.close, slave)
|
|
self.radio = SimpleNamespace(fd=slave, path=os.ttyname(slave),
|
|
expected=owner.identity(slave), board='rak4631')
|
|
self.service = owner.Owner.__new__(owner.Owner)
|
|
self.service.radio = self.radio
|
|
self.service.uid = self.service.gid = 53002
|
|
self.service.home = slave # Job is replaced; no process consumes it.
|
|
self.service.tools = {'esptool': 456, 'rnode': 789}
|
|
self.service.groups = None
|
|
self.service.custody = lambda: None
|
|
self.service.enter = lambda: None
|
|
self.service.admission_end = owner.now() + 2_000_000_000
|
|
self.service.image = None
|
|
self.args = {'action': 'write', 'board': 'rak4631', 'band': '1', 'nonce': 'ab' * 16}
|
|
|
|
def load(self, raw):
|
|
self.service.operation('image', {'size': len(raw), 'sha256': hashlib.sha256(raw).hexdigest()})
|
|
self.addCleanup(self.service.clear_image)
|
|
self.service.operation('image-chunk', {'data': raw.hex()})
|
|
|
|
def test_owner_grants_only_sealed_image_and_fixed_destination(self):
|
|
self.load(firmware())
|
|
with tempfile.TemporaryFile() as target:
|
|
def execute(groups, argv, fds, env, uid, gid, tool, *args):
|
|
self.assertEqual(argv, ['/usr/local/bin/archy-esptool', '--archy-uf2-owned', self.args['nonce']])
|
|
self.assertEqual(tool, 456)
|
|
self.assertEqual(set(fds), {3, 4, 5, 6})
|
|
import fcntl
|
|
self.assertTrue(fcntl.fcntl(fds[5], fcntl.F_GET_SEALS) & fcntl.F_SEAL_WRITE)
|
|
self.assertEqual(uf2.copy_image(fds[5], fds[6])['board'], 'rak4631')
|
|
with patch.object(uf2, 'open_destination', return_value=os.dup(target.fileno())) as destination, patch.object(owner, 'Job', side_effect=execute):
|
|
self.assertTrue(self.service.operation('run', self.args))
|
|
destination.assert_called_once_with(self.radio.path, self.args['nonce'])
|
|
target.seek(0); self.assertEqual(target.read(), firmware())
|
|
|
|
def test_other_actions_wrong_board_and_malformed_image_do_not_touch_drive(self):
|
|
self.load(b'invalid firmware')
|
|
with patch.object(uf2, 'open_destination') as destination, patch.object(owner, 'Job') as job:
|
|
for action in ('install', 'verify', 'write-slow', 'capabilities', 'write'):
|
|
with self.subTest(action=action), self.assertRaises(ValueError):
|
|
self.service.operation('run', {**self.args, 'action': action})
|
|
with self.assertRaises(ValueError):
|
|
self.service.operation('run', {**self.args, 'board': 'heltecv4'})
|
|
destination.assert_not_called(); job.assert_not_called()
|
|
|
|
def test_replaced_serial_path_does_not_grant_volume(self):
|
|
self.load(firmware())
|
|
self.radio.expected[1] += 1
|
|
with patch.object(uf2, 'open_destination') as destination, self.assertRaisesRegex(ValueError, 'appointed radio'):
|
|
self.service.operation('run', self.args)
|
|
destination.assert_not_called()
|
|
|
|
|
|
class VolumeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.usb = self.root / 'devices/1-2'
|
|
self.usb.mkdir(parents=True)
|
|
for name, value in [('idVendor', '239a'), ('idProduct', '0029'), ('serial', 'board-unique')]:
|
|
(self.usb / name).write_text(value + '\n')
|
|
(self.usb / 'tty/ttyACM0').mkdir(parents=True)
|
|
(self.usb / 'block/sda').mkdir(parents=True)
|
|
tty = self.root / 'tty/ttyACM0'; tty.mkdir(parents=True)
|
|
(tty / 'device').symlink_to(self.usb / 'tty/ttyACM0')
|
|
(self.root / 'block').mkdir()
|
|
(self.root / 'block/sda').symlink_to(self.usb / 'block/sda')
|
|
self.volume = self.root / 'RAK volume'; self.volume.mkdir()
|
|
device = self.volume.stat().st_dev
|
|
self.number = f'{os.major(device)}:{os.minor(device)}'
|
|
(self.usb / 'block/sda/dev').write_text(self.number)
|
|
(self.volume / 'INFO_UF2.TXT').write_text('Board-ID: WisBlock-RAK4631-Board\nSoftDevice: S140 6.1.1\n')
|
|
self.mountinfo = self.root / 'mountinfo'
|
|
self.mountinfo.write_text(self.mount_line())
|
|
mapping = {'/sys/class/tty': self.root / 'tty', '/sys/class/block': self.root / 'block',
|
|
'/proc/self/mountinfo': self.mountinfo}
|
|
self.paths = patch.object(uf2, 'Path', side_effect=lambda value: mapping.get(str(value), Path(value)))
|
|
self.paths.start(); self.addCleanup(self.paths.stop)
|
|
|
|
def mount_line(self, number=None):
|
|
path = str(self.volume).replace(' ', '\\040')
|
|
return f'1 0 {number or self.number} / {path} rw - vfat /dev/sda rw\n'
|
|
|
|
def test_same_usb_volume_only_and_exclusive_creation(self):
|
|
nonce = 'ab' * 16
|
|
fd = uf2.open_destination('/dev/ttyACM0', nonce)
|
|
os.close(fd)
|
|
with self.assertRaises(FileExistsError):
|
|
uf2.open_destination('/dev/ttyACM0', nonce)
|
|
self.assertTrue((self.volume / ('archy-' + nonce + '.uf2')).exists())
|
|
|
|
def test_no_label_search_or_unrelated_volume_fallback(self):
|
|
self.mountinfo.write_text(self.mount_line('999:999'))
|
|
with self.assertRaisesRegex(ValueError, 'mount the RAK4631'):
|
|
uf2.open_boot_volume('/dev/ttyACM0')
|
|
|
|
def test_ambiguous_mount_wrong_board_and_symlink_are_refused(self):
|
|
self.mountinfo.write_text(self.mount_line() * 2)
|
|
with self.assertRaises(ValueError): uf2.open_boot_volume('/dev/ttyACM0')
|
|
self.mountinfo.write_text(self.mount_line())
|
|
info = self.volume / 'INFO_UF2.TXT'
|
|
info.write_text('Board-ID: another-board\n')
|
|
with self.assertRaises(ValueError): uf2.open_boot_volume('/dev/ttyACM0')
|
|
other = self.root / 'elsewhere'; other.write_text('Board-ID: WisBlock-RAK4631-Board\n')
|
|
info.unlink(); info.symlink_to(other)
|
|
with self.assertRaises(OSError): uf2.open_boot_volume('/dev/ttyACM0')
|
|
|
|
def test_usb_identity_change_is_refused(self):
|
|
original = uf2.usb_identity
|
|
initial = original(self.usb)
|
|
with patch.object(uf2, 'usb_identity', side_effect=[initial, (*initial[:4], 'replacement')]), self.assertRaisesRegex(ValueError, 'changed'):
|
|
uf2.open_boot_volume('/dev/ttyACM0')
|
|
|
|
def test_only_native_bootloader_is_admitted(self):
|
|
for path in ('/dev/ttyUSB0', '/dev/serial/by-id/a', '/tmp/ttyACM0'):
|
|
with self.subTest(path=path), self.assertRaises(ValueError): uf2.open_boot_volume(path)
|
|
(self.usb / 'idProduct').write_text('8029')
|
|
with self.assertRaisesRegex(ValueError, 'bootloader mode'): uf2.open_boot_volume('/dev/ttyACM0')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|