203 lines
9.5 KiB
Python
203 lines
9.5 KiB
Python
"""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()
|