"""Synthetic RNS 1.3.5 serial/crypto fixtures; never opens a real tty.""" import hashlib import os import tempfile import unittest from types import SimpleNamespace from unittest.mock import patch import archy_rnodeconf as wrapper class WrapperTests(unittest.TestCase): def test_owned_daemon_keeps_destination_without_reading_node_seed(self): import asyncio from reticulum_daemon import ReticulumDaemon from archy_rns_identity import rns_private_blob, lxmf_destination_hash seed=bytes(range(32)) with patch.object(ReticulumDaemon,'_read_seed',side_effect=AssertionError('node seed must not enter helper')): daemon=ReticulumDaemon(SimpleNamespace(owned_identity=rns_private_blob(seed))) try: self.assertIsNone(daemon.seed) self.assertEqual(daemon.dest_hash_hex,lxmf_destination_hash(seed).hex()) finally: daemon.loop.close();asyncio.set_event_loop(None) def test_retained_keys_are_sealed_read_only_snapshot_not_live_path_reads(self): import builtins import fcntl import json from pathlib import Path import archy_owned_keys with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) (root/'firmware').mkdir(); (root/'trusted_keys').mkdir() signing = root/'firmware/signing.key' signing.write_bytes(b'live-path-must-not-be-used') snapshot = {'firmware/signing.key': b'original-signing-key'.hex(), 'firmware/device.key': b'original-device-key'.hex(), 'trusted_keys/retained.pubkey': b'original-trust'.hex()} fd = os.memfd_create('fixture-keys', os.MFD_ALLOW_SEALING) os.write(fd,json.dumps(snapshot).encode()); os.lseek(fd,0,0) fcntl.fcntl(fd,fcntl.F_ADD_SEALS,fcntl.F_SEAL_WRITE|fcntl.F_SEAL_GROW|fcntl.F_SEAL_SHRINK|fcntl.F_SEAL_SEAL) originals = builtins.open, os.path.isfile, os.listdir try: archy_owned_keys.install(SimpleNamespace(CNF_DIR=str(root),TK_DIR=str(root/'trusted_keys')),fd) with open(signing,'rb') as key: self.assertEqual(key.read(),b'original-signing-key') self.assertTrue(os.path.isfile(root/'firmware/device.key')) with self.assertRaises(ValueError): open(signing,'wb') self.assertEqual(os.listdir(root/'trusted_keys'),['retained.pubkey']) finally: builtins.open, os.path.isfile, os.listdir = originals os.close(fd) self.assertEqual(signing.read_bytes(),b'live-path-must-not-be-used') def test_owned_serial_uses_same_open_description_without_path_open(self): import fcntl import termios import serial import serial.tools.list_ports import archy_owned_serial master, slave = os.openpty() fcntl.ioctl(slave, termios.TIOCEXCL) original, ports = serial.Serial, serial.tools.list_ports.comports try: archy_owned_serial.install('/dev/ttyUSB-does-not-exist', slave) with patch.object(os, 'open', side_effect=AssertionError('pathname reopen forbidden')): port = serial.Serial('/dev/ttyUSB-does-not-exist', 115200, timeout=.2) self.assertEqual(os.fstat(port.fd).st_ino, os.fstat(slave).st_ino) port.write(b'owned-fd') self.assertEqual(os.read(master, 8), b'owned-fd') os.write(master, b'reply') self.assertEqual(port.read(5), b'reply') with self.assertRaises(serial.SerialException): serial.Serial('/dev/null', 115200) port.close() self.assertTrue(os.isatty(slave), 'borrower close must retain appointment') finally: serial.Serial, serial.tools.list_ports.comports = original, ports os.close(master); os.close(slave) def test_owned_serial_refuses_nonexclusive_and_non_tty_fds(self): import archy_owned_serial master, slave = os.openpty() try: with self.assertRaises(ValueError): archy_owned_serial.install('/dev/ttyUSB0', slave) with tempfile.TemporaryFile() as file: with self.assertRaises(ValueError): archy_owned_serial.install('/dev/ttyUSB0', file.fileno()) finally: os.close(master); os.close(slave) def test_frozen_writer_uses_itself_not_downloaded_python(self): import sys args = [sys.executable, "/cache/release/esptool.py", "--chip", "esp32s3", "write_flash", "0x0", "/cache/firmware.bin"] with patch.object(sys, "frozen", True, create=True): self.assertEqual(wrapper.packaged_flasher(args), [sys.executable, "--archy-esptool", "--chip", "esp32s3", "write-flash", "0x0", "/cache/firmware.bin"]) for other in (["python3", "other.py"], ["unzip", "archive.zip"]): self.assertEqual(wrapper.packaged_flasher(other), other) def test_exit_zero_without_writer_is_failure(self): for message in ("Could not find specified port", "Device signature validated", "RNode did not respond"): def main(): print(message) raise SystemExit(0) with self.assertRaises(ValueError): wrapper.install(SimpleNamespace(main=main), "/dev/ttyACM0") def test_actual_writer_exit_is_required(self): import subprocess def main(): subprocess.call(["esptool", "write_flash", "0x0", "fixture.bin"]) for code in (0, 1): with patch.object(subprocess, "call", return_value=code): if code: with self.assertRaises(ValueError): wrapper.install(SimpleNamespace(main=main), "/dev/ttyACM0") else: wrapper.install(SimpleNamespace(main=main), "/dev/ttyACM0") def test_arbitrary_and_missing_targets_rejected(self): with tempfile.TemporaryDirectory() as directory: alias = os.path.join(directory, "radio") os.symlink("/dev/null", alias) for path in (alias, "/dev/null", "/dev/ttyACM99999999", "--help"): with self.assertRaises((ValueError, OSError)): wrapper.serial_target(path) class SyntheticSerialTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.home = tempfile.TemporaryDirectory() # Upstream creates its cache on import. Keep it outside real app/home data. with patch.dict(os.environ, HOME=cls.home.name): from RNS.Utilities import rnodeconf cls.rns = rnodeconf from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.primitives import serialization cls.key = rsa.generate_private_key(public_exponent=65537, key_size=1024) pub = cls.key.public_key().public_bytes(serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo) cls.rns.known_keys = [["LOCAL", pub.hex()]] @classmethod def tearDownClass(cls): cls.home.cleanup() def radio(self, *, corrupt_signature=False, mismatch=False, missing=False): from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import padding r = self.rns eeprom = bytearray(256) eeprom[:11] = bytes([0xC3, 0xC8, 1, 0, 0, 0, 7, 0, 0, 0, 1]) checksum = hashlib.md5(eeprom[:11]).digest() eeprom[11:27] = checksum eeprom[27:155] = self.key.sign(checksum, padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH), hashes.SHA256()) eeprom[155] = 0x73 if corrupt_signature: eeprom[28] ^= 1 def frame(command, data): return bytes([0xC0, command]) + r.KISS.escape(bytes(data)) + b'\xc0' stream = frame(0x08, [0x46]) + frame(0x50, [1, 86]) stream += frame(0x60, b'\x01' + b'a' * 32) if not missing: stream += frame(0x60, b'\x02' + (b'b' if mismatch else b'a') * 32) stream += frame(0x51, eeprom) class Serial: name = "synthetic-only" def __init__(self): self.data = bytearray(stream) @property def is_open(self): return bool(self.data) @property def in_waiting(self): return len(self.data) def read(self, count): result = bytes(self.data[:count]) del self.data[:count] return result node = r.RNode(Serial()) node.readLoop() # exercise upstream's actual KISS unescaping/parser node.parse_eeprom() # actual RSA verification, not a boolean mock return node def test_serial_eeprom_signature_and_firmware_are_verified(self): result = wrapper.verified_state(self.radio(), "heltecv4") self.assertEqual(result["firmware"], "1.86") self.assertEqual(result["sha256"], (b'a' * 32).hex()) def test_corrupt_signature_missing_hash_mismatch_wrong_board_and_disconnect(self): for options in ({"corrupt_signature": True}, {"mismatch": True}, {"missing": True}): with self.subTest(options=options), self.assertRaises(ValueError): wrapper.verified_state(self.radio(**options), "heltecv4") with self.assertRaises(ValueError): wrapper.verified_state(self.radio(), "heltecv3") with self.assertRaises(ValueError): wrapper.verified_state(None, "heltecv4") if __name__ == "__main__": unittest.main()