diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index 5cdcf1452..ccc67d86f 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -582,6 +582,8 @@ jobs: python3 -B test/test_host_kiss_infrastructure.py -v python3 -B test/test_host_kiss.py -v python3 -B test/test_host_kiss_usb.py -v + python3 -B test/test_host_kiss_ingress.py -v + python3 -B test/test_host_kiss_capabilities.py -v python3 -B test/test_ethernet_kiss.py -v python3 -B test/test_ethernet_radio.py -v python3 -B test/test_companion_ethernet.py -v diff --git a/build_legacy.sh b/build_legacy.sh index 86808509b..137780c32 100755 --- a/build_legacy.sh +++ b/build_legacy.sh @@ -4809,6 +4809,25 @@ sys.exit(0 if required else 3) esac } +requires_host_kiss_code() { + local env_name=$1 + local pio_env_name=$2 + local env_platform=$3 + local decision_status=0 + local -a arguments=(--required --env "$pio_env_name" --platform "$env_platform") + is_kiss_modem_target "$env_name" && return 1 + if is_companion_radio_full_target "$env_name"; then + arguments+=(--full-companion) + fi + python3 scripts/host_kiss_capabilities.py "${arguments[@]}" \ + <<<"$PIO_CONFIG_JSON" || decision_status=$? + case "$decision_status" in + 0) return 0 ;; + 3) return 1 ;; + *) return 2 ;; + esac +} + write_build_capability_manifest() { local env_name=$1 local env_platform=$2 @@ -4846,6 +4865,14 @@ write_build_capability_manifest() { echo "Cannot determine the Full Companion Meshtext capability contract." >&2 return "$meshtext_status" fi + local host_kiss_status=0 + requires_host_kiss_code "$env_name" "$pio_env_name" "$env_platform" || host_kiss_status=$? + if [ "$host_kiss_status" -eq 0 ]; then + checker_args+=(--require-host-kiss) + elif [ "$host_kiss_status" -ne 1 ]; then + echo "Cannot determine the host KISS capability contract." >&2 + return "$host_kiss_status" + fi if [ "$env_platform" = "ESP32_PLATFORM" ]; then checker_args+=(--firmware-bin "${build_output_dir}/firmware.bin" --partitions "${build_output_dir}/partitions.bin") diff --git a/docs/kiss_modem_protocol.md b/docs/kiss_modem_protocol.md index 9088c6dca..af82cc938 100644 --- a/docs/kiss_modem_protocol.md +++ b/docs/kiss_modem_protocol.md @@ -36,7 +36,9 @@ If the initial radio2 change is still pending, the connection waits and sends READY when radio2 becomes available; no reconnect is needed. DATA received while waiting is discarded, so hosts should wait for the initial READY. -USB remains the normal CLI until an initial KISS FEND (`C0`). During the raw +USB remains the normal CLI until an initial KISS FEND (`C0`). A delimiter +inside an unfinished Companion frame or an active mOTA session keeps its +original protocol ownership. During the raw session, USB Companion framing and text/debug output are parked. Send the standard KISS RETURN frame (`C0 FF C0`) to leave the session, then issue CLI commands normally. A detected USB disconnect/reset also ends the session. diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index bd8fbb88a..f2bfdf487 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/examples/companion_radio/MyMesh.cpp @@ -11182,6 +11182,7 @@ void MyMesh::checkCLIRescueCmd() { } unsigned budget = sizeof(cli_command); while (budget-- && input.available()) { + if (mesh::hostkiss::usbInputReserved(input.peek())) return; char c = input.read(); output.print(c); // echo if (c == '\r' || c == '\n') { diff --git a/examples/companion_radio/main.cpp b/examples/companion_radio/main.cpp index aaad11ab5..9e71a8022 100644 --- a/examples/companion_radio/main.cpp +++ b/examples/companion_radio/main.cpp @@ -1113,6 +1113,15 @@ static void resetUsbTerminalHostSession() { // Park this one Companion transport; BLE/WiFi and the primary mesh stay live. static bool serviceCompanionHostKissUsb() { #if MESH_ENABLE_HOST_KISS + // A FEND inside an already-started Companion frame is payload, not a + // request to change protocols. The idle ingress gate handles the next one. + if (!mesh::hostkiss::usbActive()) { + if (usb_serial_interface.isReceivingFrame()) return false; +#if COMPANION_FEATURE_USB_MOTA_SOURCE + if (usb_mota_mode + || mesh::ota::OtaContext::serialFolderSource().hasPendingResponse()) return false; +#endif + } const bool claimed = mesh::hostkiss::serviceUsb(); if (claimed && mesh::hostkiss::usbActive() && !host_kiss_usb_parked) { host_kiss_usb_restore_enabled = usb_serial_interface.isEnabled(); @@ -1320,6 +1329,7 @@ static void serviceUsbTerminal() { bool blank_input = false; int pending = usb_input.available(); while (pending-- > 0) { + if (mesh::hostkiss::usbInputReserved(usb_input.peek())) return; // A leftover CR/LF (for example after mOTA shutdown) must not hide a // following frame or cause a banner to poison its first binary reply. if (usb_terminal_line_len == 0 && !usb_terminal_discard_line) { @@ -1488,7 +1498,10 @@ static void serviceUsbLoggingOwnership(bool logging_enabled) { // the stream quarantined forever. Let its bounded consumer drain instead. Stream& input = mesh::usbCompanionPort(); int pending = response_pending ? 0 : input.available(); - while (pending-- > 0) input.read(); + while (pending-- > 0) { + if (mesh::hostkiss::usbInputReserved(input.peek())) return; + input.read(); + } usb_logging_network_parked = false; usb_serial_interface.setPassthroughMode( the_mesh.isTerminalMode() || response_pending); @@ -1610,7 +1623,10 @@ static void expireUsbBinaryStartupProbeBeforeDispatch() { // already queued for that expired attempt so they cannot begin a new probe. Stream& usb_input = mesh::usbCompanionPort(); int pending = usb_input.available(); - while (pending-- > 0) usb_input.read(); + while (pending-- > 0) { + if (mesh::hostkiss::usbInputReserved(usb_input.peek())) return; + usb_input.read(); + } enterUsbTerminalMode(); } #endif @@ -4141,6 +4157,11 @@ void setup() { // keep frames intact and pace the contact stream when the host is slow usb_serial_interface.enableFlowControl(true); usb_serial_interface.setReceiveFrameCheck(acceptUsbBinaryStartupFrame); +#if MESH_ENABLE_HOST_KISS + usb_serial_interface.setIdleReceiveCheck([](int next_byte) { + return !mesh::hostkiss::usbInputReserved(next_byte); + }); +#endif #if defined(ESP32) && defined(ARDUINO_USB_MODE) && ARDUINO_USB_MODE == 1 \ && defined(ARDUINO_USB_CDC_ON_BOOT) && ARDUINO_USB_CDC_ON_BOOT // The ESP32 USB-Serial-JTAG peripheral (HWCDC) has no DTR concept at all. diff --git a/examples/simple_repeater/main.cpp b/examples/simple_repeater/main.cpp index f74103a43..21a3a6a7a 100644 --- a/examples/simple_repeater/main.cpp +++ b/examples/simple_repeater/main.cpp @@ -526,6 +526,10 @@ static void __attribute__((noinline)) serviceCommandInterfaces() { bool overlong_line_complete = false; size_t read_budget = 256; while (usb_ready && read_budget-- > 0 && console.available()) { +#if MESH_ENABLE_HOST_KISS + // Bytes may arrive after the loop-top KISS poll. Preserve its first FEND. + if (mesh::hostkiss::usbInputReserved(console.peek())) break; +#endif char c = console.read(); if (c == '\n') continue; console.print(c); diff --git a/examples/simple_room_server/main.cpp b/examples/simple_room_server/main.cpp index 06c64d512..a6498e285 100644 --- a/examples/simple_room_server/main.cpp +++ b/examples/simple_room_server/main.cpp @@ -392,6 +392,10 @@ void loop() { len = 0; } while (usb_ready && console.available() && len < sizeof(command)-1) { +#if MESH_ENABLE_HOST_KISS + // Bytes may arrive after the loop-top KISS poll. Preserve its first FEND. + if (mesh::hostkiss::usbInputReserved(console.peek())) break; +#endif char c = console.read(); if (c != '\n') { command[len++] = c; diff --git a/examples/simple_sensor/main.cpp b/examples/simple_sensor/main.cpp index 51b671112..0a98f0f3c 100644 --- a/examples/simple_sensor/main.cpp +++ b/examples/simple_sensor/main.cpp @@ -375,6 +375,10 @@ void loop() { len = 0; } while (usb_ready && console.available() && len < sizeof(command)-1) { +#if MESH_ENABLE_HOST_KISS + // Bytes may arrive after the loop-top KISS poll. Preserve its first FEND. + if (mesh::hostkiss::usbInputReserved(console.peek())) break; +#endif char c = console.read(); if (c != '\n') { command[len++] = c; diff --git a/scripts/check_firmware_capabilities.py b/scripts/check_firmware_capabilities.py index 431d358b9..51a5e68a9 100644 --- a/scripts/check_firmware_capabilities.py +++ b/scripts/check_firmware_capabilities.py @@ -12,6 +12,10 @@ import struct import zipfile import zlib from rp2040_application import read_application as read_rp2040_application +from host_kiss_capabilities import (CAPABILITY as HOST_KISS_CAPABILITY, + USB_CAPABILITY as HOST_KISS_USB_CAPABILITY, TCP_CAPABILITY as HOST_KISS_TCP_CAPABILITY, + COMMAND_MARKERS as HOST_KISS_COMMAND_MARKERS, TCP_MARKER as HOST_KISS_TCP_MARKER, + evidence as host_kiss_evidence) from package_meshtext_license import (CAPABILITY as MESHTEXT_CAPABILITY, meshtext_evidence, notice_bytes as meshtext_notice_bytes, notice_record as meshtext_notice_record) @@ -81,6 +85,10 @@ def parse_args() -> argparse.Namespace: "--require-meshtext", action="store_true", help="Require linked Meshtext encode/decode and Companion raw-frame transport implementations.", ) + parser.add_argument( + "--require-host-kiss", action="store_true", + help="Require linked KISS framing/Dispatcher transport and packaged USB/TCP service commands.", + ) parser.add_argument("--firmware-bin", type=Path) parser.add_argument("--partitions", type=Path) parser.add_argument("--dfu-package", type=Path) @@ -231,12 +239,34 @@ def main() -> int: "evidence": "Meshtext encode/decode, Companion send/receive, and Dispatcher raw-frame queue implementations", "present": False, "source": "linked ELF symbols"}) application = b"" - if args.expect_application: + if args.expect_application or args.require_host_kiss: try: application = read_application(args) except (OSError, ValueError, KeyError, zipfile.BadZipFile) as exc: print(f"capability check: cannot read packaged application: {exc}", file=sys.stderr) malformed = True + host_kiss_symbols = [] + host_kiss_verified = False + if args.require_host_kiss: + try: + host_kiss_symbols = host_kiss_evidence(args.image) + except (OSError, ValueError) as exc: + print(f"capability check: cannot qualify linked host KISS code: {exc}", file=sys.stderr) + malformed = True + checks.append({"capability": HOST_KISS_CAPABILITY, + "evidence": "; ".join(host_kiss_symbols) if host_kiss_symbols else + "executable KISS frame decoder/encoder and Dispatcher raw-frame queue", + "present": bool(host_kiss_symbols), "source": "linked ELF symbols"}) + usb_present = all(marker in application for marker in HOST_KISS_COMMAND_MARKERS) + checks.append({"capability": HOST_KISS_USB_CAPABILITY, + "evidence": "; ".join(marker.decode("ascii") for marker in HOST_KISS_COMMAND_MARKERS), + "present": usb_present, "source": "packaged application"}) + tcp_present = args.platform != "ESP32_PLATFORM" or HOST_KISS_TCP_MARKER in application + if args.platform == "ESP32_PLATFORM": + checks.append({"capability": HOST_KISS_TCP_CAPABILITY, + "evidence": HOST_KISS_TCP_MARKER.decode("ascii"), + "present": tcp_present, "source": "packaged application"}) + host_kiss_verified = bool(host_kiss_symbols) and usb_present and tcp_present expectations = [(value, image, "linked image") for value in args.expect] expectations += [(value, application, "packaged application") for value in args.expect_application] gzip_assets = linked_gzip_assets(image) if args.expect_gzip else [] @@ -269,6 +299,11 @@ def main() -> int: # Neither a requested capability nor CLI strings prove the optional # codec and its complete Companion transport survived linking. capabilities = [value for value in capabilities if value != MESHTEXT_CAPABILITY] + if not host_kiss_verified: + capabilities = [value for value in capabilities if value not in + (HOST_KISS_CAPABILITY, HOST_KISS_USB_CAPABILITY, HOST_KISS_TCP_CAPABILITY)] + if args.platform != "ESP32_PLATFORM": + capabilities = [value for value in capabilities if value != HOST_KISS_TCP_CAPABILITY] missing = [check for check in checks if not check["present"]] ota_methods = stable_unique([ check["capability"].removeprefix("ota.update.") diff --git a/scripts/host_kiss_capabilities.py b/scripts/host_kiss_capabilities.py new file mode 100644 index 000000000..1ef819702 --- /dev/null +++ b/scripts/host_kiss_capabilities.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +"""Qualify the optional host KISS service from effective flags and linked code.""" +from __future__ import annotations + +import argparse +import json +import os +from pathlib import Path +import re +import shlex +import sys + +from package_meshtext_license import linked_symbols + +CAPABILITY = "radio2.kiss.host" +USB_CAPABILITY = "radio2.kiss.usb" +TCP_CAPABILITY = "radio2.kiss.tcp" +COMMAND_MARKERS = (b"Error: use set kiss on|off", + b"Error: use set kiss.transport usb|tcp", + b"USB: send KISS FEND to start") +TCP_MARKER = b"station TCP 5004" + + +def evidence(path): + # LTO can inline the transport callbacks. Require executable framing and + # the bounded Dispatcher queue shared by the actual RX/TX implementation. + # A declaration, an undefined reference, or a CLI help string is not code. + symbols = linked_symbols(path) + required = ("_ZN4mesh8hostkiss5Codec4feedE", + "_ZN4mesh8hostkiss5Codec11queueOutputE", + "_ZN4mesh10Dispatcher22queueForeignRadioFrameE") + selected = [] + for operation in required: + matches = [name for name in symbols if name.startswith(operation)] + if not matches: + return [] + selected.append(matches[0]) + return selected + + +def operations(value): + words = iter(shlex.split(value if isinstance(value, str) else "\n".join(map(str, value)))) + result = [] + for word in words: + if word in ("-D", "-U"): + word += next(words, "") + match = re.fullmatch(r"-([DU])(\w+)(?:=(.*))?", word) + if match: + result.append(match.groups()) + return result + + +def required(config, env, platform, full_companion, overlay="", unflags=""): + if platform not in ("ESP32_PLATFORM", "NRF52_PLATFORM"): + return False + rows = {name: dict(options) for name, options in config} + options = rows.get("env:" + env, {}) + flags = operations(options.get("build_flags", [])) + operations(overlay) + removed = set(operations(options.get("build_unflags", [])) + operations(unflags)) + flags = [flag for flag in flags if flag not in removed + and not (flag[0] == "D" and ("D", flag[1], None) in removed)] + defines = {name: "1" if value is None else value + for kind, name, value in flags if kind == "D"} + # PlatformIO emits surviving -U after CPPDEFINES, so -U always wins. + for kind, name, _value in flags: + if kind == "U": + defines.pop(name, None) + + def enabled(name, default=False, seen=()): + if name not in defines: + return default + if name in seen: + return None + value = defines[name].strip().strip("() ") + if value in ("false", "true"): + return value == "true" + if re.fullmatch(r"[A-Za-z_]\w*", value): + return enabled(value, False, (*seen, name)) + number = re.fullmatch(r"([+-]?(?:0[xX][0-9a-fA-F]+|[0-9]+))[uUlL]*", value) + if number: + value = number[1] + base = 16 if "x" in value.lower() else 8 if re.fullmatch(r"[+-]?0[0-7]+", value) else 10 + return int(value, base) != 0 + return None + + # Match HostKissConfig.h's hard exclusion even with an explicit enable. + if enabled("MESH_PRIMARY_ESPNOW") is True: + return False + if "MESH_ENABLE_HOST_KISS" in defines: + return enabled("MESH_ENABLE_HOST_KISS") is not False + return enabled("COMPANION_RADIO_FULL", full_companion) is not False + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--required", action="store_true") + parser.add_argument("--env", required=True) + parser.add_argument("--platform", required=True) + parser.add_argument("--full-companion", action="store_true") + args = parser.parse_args() + try: + selected = required(json.load(sys.stdin), args.env, args.platform, args.full_companion, + os.environ.get("PLATFORMIO_BUILD_FLAGS", ""), + os.environ.get("PLATFORMIO_BUILD_UNFLAGS", "")) + except (OSError, ValueError, TypeError) as error: + parser.exit(2, f"Cannot determine host KISS capability contract: {error}\n") + return 0 if selected else 3 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/helpers/ArduinoSerialInterface.cpp b/src/helpers/ArduinoSerialInterface.cpp index 4e54a42a6..4c92ae003 100644 --- a/src/helpers/ArduinoSerialInterface.cpp +++ b/src/helpers/ArduinoSerialInterface.cpp @@ -307,6 +307,10 @@ size_t ArduinoSerialInterface::checkRecvFrame(uint8_t dest[]) { // host must not keep the radio/watchdog loop trapped in this parser. int pending = _serial->available(); while (pending-- > 0) { + // Input can arrive after the role's initial protocol-owner poll. Leave a + // reserved first byte for that owner without consuming its framing. + if (_state == RECV_STATE_IDLE && _idle_receive_check != nullptr + && !_idle_receive_check(_serial->peek())) return 0; int c = _serial->read(); if (c < 0) break; _last_rx_byte_ms = millis(); diff --git a/src/helpers/ArduinoSerialInterface.h b/src/helpers/ArduinoSerialInterface.h index 532956735..046fb6bc0 100644 --- a/src/helpers/ArduinoSerialInterface.h +++ b/src/helpers/ArduinoSerialInterface.h @@ -10,6 +10,9 @@ public: // Optional protocol-owner gate, checked before a completed frame is exposed // or recorded as client activity. Other transports retain their old behavior. typedef bool (*ReceiveFrameCheck)(uint32_t completed_at); + // Before reading an idle byte, another protocol may reserve it in-place. + // Length-prefixed frame bodies bypass this gate, including opaque 0xC0. + typedef bool (*IdleReceiveCheck)(int next_byte); private: static constexpr uint8_t TX_QUEUE_SIZE = 4; @@ -46,6 +49,7 @@ private: const char* _secondaryControlSequence; ConnectedCheck _conn_check; ReceiveFrameCheck _receive_frame_check; + IdleReceiveCheck _idle_receive_check; uint8_t rx_buf[MAX_FRAME_SIZE]; TxFrame _tx_queue[TX_QUEUE_SIZE]; @@ -71,7 +75,7 @@ public: _has_received_frame(false), _serial(nullptr), _controlSequence(nullptr), _secondaryControlSequence(nullptr), _conn_check(nullptr), - _receive_frame_check(nullptr) {} + _receive_frame_check(nullptr), _idle_receive_check(nullptr) {} void begin(Stream& serial, const char* controlSequence = nullptr, const char* secondaryControlSequence = nullptr) { @@ -105,6 +109,8 @@ public: // Without it isConnected() assumes true, as a plain UART has no way of knowing. void setConnectedCheck(ConnectedCheck fn) { _conn_check = fn; } void setReceiveFrameCheck(ReceiveFrameCheck fn) { _receive_frame_check = fn; } + void setIdleReceiveCheck(IdleReceiveCheck fn) { _idle_receive_check = fn; } + bool isReceivingFrame() const { return _state != 0; } // Clear all protocol state owned by one host session while retaining the // monotonic completed-frame counter used by startup handoff arbitration. diff --git a/src/helpers/HostKiss.cpp b/src/helpers/HostKiss.cpp index 60bf7cbe0..ea1f6f8a3 100644 --- a/src/helpers/HostKiss.cpp +++ b/src/helpers/HostKiss.cpp @@ -522,6 +522,10 @@ bool serviceUsb() { } bool usbActive() { return usb_session; } +bool usbInputReserved(int next_byte) { + return usb_session || (settings.selection().enabled + && settings.selection().transport == Transport::Usb && next_byte == Codec::Fend); +} bool radioActive() { return current(); } void resetUsbSession() { if (usb_session) closeSession(); } diff --git a/src/helpers/HostKiss.h b/src/helpers/HostKiss.h index d83974c9a..3a1f28ce3 100644 --- a/src/helpers/HostKiss.h +++ b/src/helpers/HostKiss.h @@ -16,6 +16,9 @@ bool handleCommand(const char* command, char* reply, size_t capacity, bool local void loop(); bool serviceUsb(); bool usbActive(); +// Normal USB consumers must leave this byte for serviceUsb(), even when it +// arrives after their loop-top owner poll. Never call inside another frame. +bool usbInputReserved(int next_byte); bool radioActive(); void resetUsbSession(); void receive(const uint8_t* bytes, size_t length, uint8_t profile, @@ -29,6 +32,7 @@ inline bool handleCommand(const char*, char*, size_t, bool) { return false; } inline void loop() {} inline bool serviceUsb() { return false; } inline bool usbActive() { return false; } +inline bool usbInputReserved(int) { return false; } inline bool radioActive() { return false; } inline void resetUsbSession() {} inline void receive(const uint8_t*, size_t, uint8_t, uint32_t, float, float) {} diff --git a/test/test_host_kiss.py b/test/test_host_kiss.py index ef6feee9e..c2b02a71b 100644 --- a/test/test_host_kiss.py +++ b/test/test_host_kiss.py @@ -343,10 +343,12 @@ int main() { GuestMemoryFS fs;Radio radio;RadioProfileCLI profiles(radio);Dispatcher dispatcher; hostkiss::begin(&fs,radio,profiles,dispatcher); command("get kiss");assert(strstr(reply,"> off")); + assert(!hostkiss::usbInputReserved(0xc0)); input({0xc0,0,1,0xc0});assert(!hostkiss::serviceUsb()&&serial.available()==4); serial.input.clear(); command("set kiss on",false);assert(strstr(reply,"local connection")); command("set kiss on");assert(strstr(reply,"OK")); + assert(hostkiss::usbInputReserved(0xc0) && !hostkiss::usbInputReserved(0)); hostkiss::Settings saved;assert(saved.begin(&fs)&&saved.selection().enabled); input({'g','e','t',' '});assert(!hostkiss::serviceUsb()&&serial.available()==4); serial.input.clear(); @@ -354,6 +356,7 @@ int main() { // same connection becomes ready automatically once the tuple is stable. profiles.busy=true;input({0xc0});poll(); assert(hostkiss::usbActive()&&!hostkiss::radioActive()&&serial.output.empty()); + assert(hostkiss::usbInputReserved(0) && hostkiss::usbInputReserved(-1)); command("get kiss");assert(strstr(reply,"waiting-radio")&&strstr(reply,"pending radio settings")); fake_millis=1999;poll();assert(serial.output.empty()); profiles.busy=false;fake_millis=2000;poll(); diff --git a/test/test_host_kiss_capabilities.py b/test/test_host_kiss_capabilities.py new file mode 100644 index 000000000..299e32b31 --- /dev/null +++ b/test/test_host_kiss_capabilities.py @@ -0,0 +1,175 @@ +#!/usr/bin/env python3 +"""Require executable KISS transport and commands in uploadable Full images.""" +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +import zipfile + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) +import host_kiss_capabilities as kiss +from test_meshtext_license import elf32 + +SYMBOLS = ("_ZN4mesh8hostkiss5Codec4feedEh", + "_ZN4mesh8hostkiss5Codec11queueOutputEhPKhj", + "_ZN4mesh10Dispatcher22queueForeignRadioFrameEPKhjhmNS_20RadioProfileProtocolE") +COMMANDS = b"\0".join(kiss.COMMAND_MARKERS) + + +class HostKissCapabilitiesTest(unittest.TestCase): + def test_effective_recipe_matches_header_overrides_and_primary_radio(self): + def required(*flags, platform="ESP32_PLATFORM", companion=True, overlay="", unflags="", config_unflags=()): + config = [["env:base", [["build_flags", list(flags)], ["build_unflags", list(config_unflags)]]]] + return kiss.required(config, "base", platform, companion, overlay, unflags) + self.assertTrue(required()) + self.assertTrue(required(platform="NRF52_PLATFORM")) + self.assertFalse(required(companion=False)) + self.assertTrue(required("-DMESH_ENABLE_HOST_KISS=1", companion=False)) + self.assertFalse(required("-DMESH_ENABLE_HOST_KISS=0")) + self.assertFalse(required("-DPOLICY=0", "-DMESH_ENABLE_HOST_KISS=POLICY")) + self.assertFalse(required("-DMESH_ENABLE_HOST_KISS=(0UL)")) + self.assertTrue(required("-DMESH_ENABLE_HOST_KISS=0", overlay="-DMESH_ENABLE_HOST_KISS=1")) + self.assertFalse(required("-DMESH_ENABLE_HOST_KISS=1", overlay="-DMESH_ENABLE_HOST_KISS=0")) + self.assertTrue(required("-DMESH_ENABLE_HOST_KISS=0", overlay="-UMESH_ENABLE_HOST_KISS")) + self.assertTrue(required("-DMESH_ENABLE_HOST_KISS=0", config_unflags=["-DMESH_ENABLE_HOST_KISS"])) + self.assertTrue(required("-DMESH_ENABLE_HOST_KISS=0", unflags="-DMESH_ENABLE_HOST_KISS=0")) + self.assertFalse(required("-DMESH_PRIMARY_ESPNOW=1", "-DMESH_ENABLE_HOST_KISS=1")) + self.assertTrue(required("-DMESH_PRIMARY_ESPNOW=1", "-UMESH_PRIMARY_ESPNOW")) + self.assertFalse(required("-DMESH_PRIMARY_ESPNOW=1", "-UMESH_PRIMARY_ESPNOW", + config_unflags=["-UMESH_PRIMARY_ESPNOW"])) + self.assertTrue(required("-DMESH_PRIMARY_ESPNOW=0")) + self.assertFalse(required("-DCOMPANION_RADIO_FULL=0")) + self.assertFalse(required("-DMESH_ENABLE_HOST_KISS=1", platform="STM32_PLATFORM")) + self.assertFalse(required("-DMESH_ENABLE_HOST_KISS=1", platform="RP2040_PLATFORM")) + + def qualify(self, *, symbols=SYMBOLS, flags=6, undefined=(), zero=(), + commands=COMMANDS, platform="ESP32_PLATFORM", tcp=True, require=True): + with tempfile.TemporaryDirectory(prefix="mesh-kiss-capabilities-") as directory: + root = Path(directory) + elf32(root / "firmware.elf", symbols=symbols, flags=flags, undefined=undefined, zero=zero) + app = commands + (b"\0" + kiss.TCP_MARKER if tcp else b"") + args = [] + if platform == "NRF52_PLATFORM": + with zipfile.ZipFile(root / "firmware.zip", "w") as package: + package.writestr("manifest.json", json.dumps({"manifest": {"application": {"bin_file": "app.bin"}}})) + package.writestr("app.bin", app) + args += ["--dfu-package", str(root / "firmware.zip")] + else: + (root / "firmware.bin").write_bytes(app) + args += ["--firmware-bin", str(root / "firmware.bin")] + if require: + args.append("--require-host-kiss") + result = subprocess.run([sys.executable, str(ROOT / "scripts/check_firmware_capabilities.py"), + "--image", str(root / "firmware.elf"), "--output", str(root / "caps.json"), + "--target", "fixture", "--platform", platform, "--build-profile", "full", + "--capability", kiss.CAPABILITY, "--capability", kiss.USB_CAPABILITY, + "--capability", kiss.TCP_CAPABILITY, *args], capture_output=True, text=True) + return result, json.loads((root / "caps.json").read_text()) + + def test_packaged_usb_and_platform_specific_tcp_capabilities(self): + for platform in ("ESP32_PLATFORM", "NRF52_PLATFORM"): + with self.subTest(platform=platform): + result, manifest = self.qualify(platform=platform, tcp=platform=="ESP32_PLATFORM") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertTrue(manifest["verified"]) + self.assertIn(kiss.CAPABILITY, manifest["capabilities"]) + self.assertIn(kiss.USB_CAPABILITY, manifest["capabilities"]) + # Requested labels alone must not advertise unsupported TCP. + self.assertEqual(kiss.TCP_CAPABILITY in manifest["capabilities"], platform=="ESP32_PLATFORM") + + def test_rejects_each_missing_or_nonexecutable_method_and_command(self): + cases = [dict(symbols=("unrelated_function",)), dict(flags=4), dict(flags=2), dict(tcp=False)] + for symbol in SYMBOLS: + cases.extend((dict(symbols=tuple(item for item in SYMBOLS if item!=symbol)), + dict(undefined=(symbol,)), dict(zero=(symbol,)))) + cases += [dict(commands=b"\0".join(marker for marker in kiss.COMMAND_MARKERS if marker!=missing)) + for missing in kiss.COMMAND_MARKERS] + for options in cases: + with self.subTest(options=options): + result, manifest = self.qualify(**options) + self.assertEqual(result.returncode, 1, result.stderr) + self.assertFalse(manifest["verified"]) + self.assertFalse(set(manifest["capabilities"]) & + {kiss.CAPABILITY,kiss.USB_CAPABILITY,kiss.TCP_CAPABILITY}) + result, manifest = self.qualify(require=False) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertFalse(set(manifest["capabilities"]) & + {kiss.CAPABILITY,kiss.USB_CAPABILITY,kiss.TCP_CAPABILITY}) + + def test_actual_builder_checks_linked_code_and_packaged_application(self): + script = r''' +set -euo pipefail +pio() { printf 'unexpected PlatformIO invocation\n' >&2; return 99; } +source build.sh +PIO_CONFIG_JSON=$MESH_TEST_CONFIG +OUTPUT_DIR=$MESH_TEST_DIRECTORY/out +PIO_BUILD_DIR_OVERRIDE=$MESH_TEST_DIRECTORY/build +BUILD_PROFILE_FOR_TARGET=full +BUILD_RECIPE_SHA256=test +BUILD_CAPABILITIES=(); BUILD_EXPECTATIONS=() +BUILD_APPLICATION_EXPECTATIONS=(); BUILD_REDUCTIONS=() +requires_wireless_self_update() { return 1; } +is_xiao_qspi_canonical_repeater_build() { return 1; } +is_nrf52_sensor_ota_pair_target() { return 1; } +embedded_bootloader_artifact_target() { printf '%s\n' "$1"; } +python3() { + if [ "$1" = scripts/firmware_build_recipe.py ]; then return 0; fi + command "$MESH_TEST_PYTHON" "$@" +} +write_build_capability_manifest fixture_repeater "$MESH_TEST_PLATFORM" base fixture +''' + for platform in ("ESP32_PLATFORM", "NRF52_PLATFORM"): + for complete in (True, False): + with self.subTest(platform=platform, complete=complete), tempfile.TemporaryDirectory() as folder: + root=Path(folder);build=root/"build/base";build.mkdir(parents=True) + elf32(build/"firmware.elf", symbols=SYMBOLS if complete else ("unrelated_function",)) + app=COMMANDS+b"\0"+kiss.TCP_MARKER + (build/"firmware.bin").write_bytes(app) + with zipfile.ZipFile(build/"firmware.zip", "w") as package: + package.writestr("manifest.json", json.dumps({"manifest":{"application":{"bin_file":"app.bin"}}})) + package.writestr("app.bin", app) + config=[["env:base", [["build_flags", ["-DMESH_ENABLE_HOST_KISS=1"]]]]] + result=subprocess.run(["bash","-c",script],cwd=ROOT,capture_output=True,text=True, + env={**os.environ,"MESH_TEST_CONFIG":json.dumps(config),"MESH_TEST_PLATFORM":platform, + "MESH_TEST_DIRECTORY":str(root),"MESH_TEST_PYTHON":sys.executable, + "PLATFORMIO_BUILD_FLAGS":"","PLATFORMIO_BUILD_UNFLAGS":""}) + self.assertEqual(result.returncode,0 if complete else 1,result.stdout+result.stderr) + manifest=json.loads((root/"out/fixture.capabilities.json").read_text()) + self.assertEqual(manifest["verified"],complete) + self.assertEqual(kiss.CAPABILITY in manifest["capabilities"],complete) + + def test_builder_requires_service_only_when_effectively_enabled(self): + script = r''' +set -euo pipefail +pio() { printf 'unexpected PlatformIO invocation\n' >&2; return 99; } +source build.sh +PIO_CONFIG_JSON=$MESH_TEST_CONFIG +is_companion_radio_full_target() { [ "$MESH_TEST_COMPANION" = 1 ]; } +is_kiss_modem_target() { [ "$MESH_TEST_MODEM" = 1 ]; } +requires_host_kiss_code fixture base "$MESH_TEST_PLATFORM" +''' + for platform, companion, modem, flags, expected in ( + ("ESP32_PLATFORM", True, False, [], 0), + ("NRF52_PLATFORM", True, False, [], 0), + ("ESP32_PLATFORM", False, False, ["-DMESH_ENABLE_HOST_KISS=1"], 0), + ("NRF52_PLATFORM", False, False, ["-DMESH_ENABLE_HOST_KISS=1"], 0), + ("ESP32_PLATFORM", True, False, ["-DMESH_ENABLE_HOST_KISS=0"], 1), + ("ESP32_PLATFORM", True, False, ["-DMESH_ENABLE_HOST_KISS=1", "-DMESH_PRIMARY_ESPNOW=1"], 1), + ("ESP32_PLATFORM", True, True, ["-DMESH_ENABLE_HOST_KISS=1"], 1), + ("STM32_PLATFORM", True, False, ["-DMESH_ENABLE_HOST_KISS=1"], 1), + ("ESP32_PLATFORM", False, False, [], 1)): + with self.subTest(platform=platform,companion=companion,modem=modem,flags=flags): + config = [["env:base", [["build_flags",flags]]]] + result = subprocess.run(["bash","-c",script],cwd=ROOT,capture_output=True,text=True, + env={**os.environ,"MESH_TEST_CONFIG":json.dumps(config),"MESH_TEST_PLATFORM":platform, + "MESH_TEST_COMPANION":str(int(companion)),"MESH_TEST_MODEM":str(int(modem)), + "PLATFORMIO_BUILD_FLAGS":"","PLATFORMIO_BUILD_UNFLAGS":""}) + self.assertEqual(result.returncode,expected,result.stdout+result.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_host_kiss_infrastructure.py b/test/test_host_kiss_infrastructure.py index 8dcf6da5a..68552433a 100644 --- a/test/test_host_kiss_infrastructure.py +++ b/test/test_host_kiss_infrastructure.py @@ -24,11 +24,16 @@ FIXTURE = r''' #include static std::deque input; static bool kiss_active = false, reset_pending = false, network_pending = false; +static bool inject_late_fend=false; static unsigned reset_reads = 0, kiss_resets = 0, ascii_commands = 0; static unsigned board_loops = 0, ethernet_loops = 0, network_commands = 0; static unsigned serial_cancels = 0, terminal_drains = 0; struct Stream { - int available() const { return int(input.size()); } + int available() const { + if(inject_late_fend){inject_late_fend=false;input={0xc0,0x00,0x41,0xc0};} + return int(input.size()); + } + int peek()const{return input.empty()?-1:input.front();} int read() { assert(!input.empty() && input.front() != 0xc0); int value = input.front(); input.pop_front(); return value; @@ -49,6 +54,7 @@ namespace power { void serviceRegulatorSettings() {} } #if MESH_ENABLE_HOST_KISS namespace hostkiss { void resetUsbSession() { ++kiss_resets; kiss_active=false; } +bool usbInputReserved(int next){return kiss_active || next==0xc0;} bool serviceUsb() { if (!input.empty() && input.front()==0xc0) { input.pop_front(); kiss_active=true; @@ -127,6 +133,11 @@ int main() { assert(command[0]==0 && !command_overflow); ascii(); serviceCommandInterfaces(); assert(ascii_commands==3); #endif + // The first FEND arrives after serviceUsb() returned false this poll. + inject_late_fend=true;const unsigned prior_commands=ascii_commands; + serviceCommandInterfaces(); + assert(!kiss_active && !input.empty() && input.front()==0xc0 && ascii_commands==prior_commands); + serviceCommandInterfaces();assert(kiss_active && input.empty()); #endif } ''' diff --git a/test/test_host_kiss_ingress.py b/test/test_host_kiss_ingress.py new file mode 100644 index 000000000..394fbd2b0 --- /dev/null +++ b/test/test_host_kiss_ingress.py @@ -0,0 +1,185 @@ +#!/usr/bin/env python3 +"""Exercise late KISS delimiters through the real Companion USB parser.""" +import os +from pathlib import Path +import shutil +import subprocess +import sys +import tempfile +import unittest + +from cpp_source import body +import test_companion_usb_input as usb +from test_replay_reset_integration import extract_braced + +ROOT = Path(__file__).resolve().parents[1] +FIXTURE = r''' +#include +#include +#include +#include +static bool enabled=true, active=false; +static unsigned gate_calls=0; +static bool receiveIdle(int next) { ++gate_calls; return !enabled || (next!=0xc0 && !active); } +struct Port:Stream { + std::deque input; + unsigned reads=0; + int available()override{return int(input.size());} + int read()override{if(input.empty())return -1;auto c=input.front();input.pop_front();++reads;return c;} + int peek()override{return input.empty()?-1:input.front();} + int availableForWrite()override{return 65536;} + size_t write(uint8_t)override{return 1;} + size_t write(const uint8_t*,size_t n)override{return n;} + void push(const std::vector& bytes){input.insert(input.end(),bytes.begin(),bytes.end());} +}; +int main(){ + Port port;ArduinoSerialInterface usb;uint8_t out[MAX_FRAME_SIZE]={}; + usb.begin(port);usb.enable();usb.setIdleReceiveCheck(receiveIdle); + assert(!usb.isReceivingFrame()); + // The top-of-loop KISS owner saw an empty port. A complete KISS frame then + // arrives during mesh dispatch; the idle Companion parser must leave it all. + assert(!usb.checkRecvFrame(out)); + const std::vector kiss={0xc0,0,0x41,0xdb,0xdc,0xc0}; + port.push(kiss);assert(!usb.checkRecvFrame(out)); + assert(port.reads==0 && std::vector(port.input.begin(),port.input.end())==kiss); + port.input.clear(); + // Every possible split of an existing length-prefixed frame preserves its + // opaque payload delimiter, including one arriving just after the split. + const std::vector binary={'<',5,0,0x42,0xc0,0xdb,0,0xff}; + for(size_t split=1;split(binary.begin(),binary.begin()+split)); + assert(!usb.checkRecvFrame(out) && usb.isReceivingFrame()); + const unsigned before=gate_calls; + port.push(std::vector(binary.begin()+split,binary.end())); + assert(usb.checkRecvFrame(out)==5 && !usb.isReceivingFrame()); + assert(std::vector(out,out+5)==std::vector(binary.begin()+3,binary.end())); + assert(gate_calls==before); // no idle gate anywhere inside the frame + } + // A completed CLI RPC followed immediately by FEND dispatches the RPC once + // and leaves the complete next protocol frame untouched on the same stream. + port.push(binary);port.push(kiss); + assert(usb.checkRecvFrame(out)==5);const unsigned before=port.reads; + assert(!usb.checkRecvFrame(out) && port.reads==before && port.peek()==0xc0); + port.input.clear(); + // Off and non-USB callers keep the original framing behavior. + enabled=false;port.push(kiss);port.push(binary); + assert(usb.checkRecvFrame(out)==5 && port.input.empty()); + usb.setIdleReceiveCheck(nullptr);enabled=true;port.push(kiss);port.push(binary); + assert(usb.checkRecvFrame(out)==5 && port.input.empty()); +} +''' + + +class HostKissIngressTest(unittest.TestCase): + def test_late_fend_is_reserved_and_binary_payload_is_opaque(self): + compiler = os.environ.get('CXX') or shutil.which('g++') or shutil.which('clang++') + self.assertIsNotNone(compiler) + source = (ROOT / 'src/helpers/ArduinoSerialInterface.cpp').read_text() + gate = (' if (_state == RECV_STATE_IDLE && _idle_receive_check != nullptr\n' + ' && !_idle_receive_check(_serial->peek())) return 0;\n') + self.assertEqual(source.count(gate), 1) + for negative in (False, True): + with self.subTest(negative=negative), tempfile.TemporaryDirectory(prefix='host-kiss-ingress-') as directory: + directory = Path(directory) + (directory / 'fixture.cpp').write_text(FIXTURE, encoding='ascii') + (directory / 'transport.cpp').write_text(source.replace(gate, '') if negative else source, encoding='ascii') + flags = (['-fsanitize=address,undefined', '-fno-sanitize-recover=all', + '-fno-pie', '-no-pie'] if sys.platform.startswith('linux') else []) + result = subprocess.run([compiler, '-std=c++17', '-Wall', '-Wextra', *flags, + '-I', str(ROOT / 'test/mocks'), '-I', str(ROOT / 'src'), + '-I', str(ROOT / 'src/helpers'), str(directory / 'fixture.cpp'), + str(directory / 'transport.cpp'), '-o', str(directory / 'test')], + text=True, capture_output=True, timeout=60) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + result = subprocess.run([str(directory / 'test')], text=True, capture_output=True, timeout=10) + if negative: + self.assertNotEqual(result.returncode, 0, 'Old parser unexpectedly preserved late KISS input') + else: + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_actual_ascii_pump_leaves_late_fend_even_with_partial_input(self): + main = (ROOT / 'examples/companion_radio/main.cpp').read_text() + signatures = ( + 'static void clearUsbTerminalLine()', + 'static bool acceptUsbBinaryStartupFrame(uint32_t completed_at)', + 'static void enterUsbTerminalMode(bool show_banner', + 'static void leaveUsbTerminalMode(bool acknowledge)', + 'static void resetUsbMotaMode()', 'static void leaveUsbMotaMode(bool acknowledge)', + 'static bool enterUsbMotaMode(mesh::UsbMotaEntryOrigin origin)', + 'static void serviceUsbMota()', 'static bool serviceUsbMotaResponseBoundary()', + 'static void resetUsbTerminalHostSession()', 'static void serviceUsbTerminal()', + 'static void expireUsbBinaryStartupProbeBeforeDispatch()', + ) + functions = '\n'.join(extract_braced(main, signature) for signature in signatures) + functions = functions.replace('static void enterUsbTerminalMode(bool show_banner)', + 'static void enterUsbTerminalMode(bool show_banner = true)') + functions = functions.replace('mesh::hostkiss::usbInputReserved(', 'reserveTestInput(') + accessor = extract_braced(usb.CONTEXT, 'static SerialMotaSource& serialFolderSource()') + declarations = ('static bool reserve_kiss=false;\n' + 'static bool reserveTestInput(int next){return reserve_kiss && next==0xc0;}\n') + checks = r""" + // An expired Binary probe must not drain FEND or print its ASCII banner. + reset();reserve_kiss=true;leaveUsbTerminalMode(false); + usb_binary_startup_probe.start(g_mock_millis,0);g_mock_millis+=1001; + input.bytes={0xc0,0,0x41,0xc0};const auto expired_output=input.output; + expireUsbBinaryStartupProbeBeforeDispatch(); + assert(input.bytes.size()==4 && input.peek()==0xc0 && input.output==expired_output); + assert(!the_mesh.terminal && usb_binary_startup_probe.isActive()); + for(unsigned mode=0;mode<3;++mode){ + reset();reserve_kiss=true; + if(mode==1){input.push("prefix");serviceUsbTerminal();} + if(mode==2)usb_terminal_discard_line=true; + const auto output_before=input.output; + const std::vector wire={0xc0,0,'v','e','r','\r',0xc0}; + for(auto byte:wire)input.bytes.push_back(byte); + serviceUsbTerminal(); + assert(std::vector(input.bytes.begin(),input.bytes.end())==wire); + assert(input.output==output_before && the_mesh.commands.empty()); + } + reserve_kiss=false; +""" + template = usb.HARNESS.replace('@FUNCTIONS@', declarations + functions) + template = template.replace('@SOURCE_ACCESSOR@', accessor).replace('int main() {', 'int main() {' + checks) + template = usb.bind_production_input_constants(template) + guard = ' if (reserveTestInput(usb_input.peek())) return;\n' + self.assertEqual(template.count(guard), 2) + for negative in (False, True): + with self.subTest(negative=negative), tempfile.TemporaryDirectory(prefix='host-kiss-ascii-') as directory: + directory = Path(directory) + (directory / 'fixture.cpp').write_text(template.replace(guard, '') if negative else template, encoding='ascii') + result = subprocess.run(['g++', '-std=c++17', '-fsanitize=address,undefined', + '-fno-sanitize-recover=all', '-fno-pie', '-no-pie', + '-I', str(ROOT / 'test/mocks'), '-I', str(ROOT / 'src'), + str(directory / 'fixture.cpp'), str(ROOT / 'src/helpers/ArduinoSerialInterface.cpp'), + str(ROOT / 'src/helpers/ota/MotaSourceSerial.cpp'), '-o', str(directory / 'test')], + text=True, capture_output=True, timeout=60) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + result = subprocess.run([str(directory / 'test')], text=True, capture_output=True, timeout=15) + if negative: + self.assertNotEqual(result.returncode, 0, 'Old ASCII pump unexpectedly preserved late FEND') + else: + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_every_raw_usb_consumer_defers_reserved_input(self): + main = (ROOT / 'examples/companion_radio/main.cpp').read_text() + setup = body(main, 'void setup()') + self.assertIn('usb_serial_interface.setIdleReceiveCheck', setup) + self.assertIn('return !mesh::hostkiss::usbInputReserved(next_byte)', setup) + for signature in ('static void serviceUsbTerminal()', + 'static void serviceUsbLoggingOwnership(bool logging_enabled)', + 'static void expireUsbBinaryStartupProbeBeforeDispatch()'): + self.assertIn('mesh::hostkiss::usbInputReserved(', body(main, signature)) + rescue = body((ROOT / 'examples/companion_radio/MyMesh.cpp').read_text(), 'void MyMesh::checkCLIRescueCmd()') + self.assertLess(rescue.index('mesh::hostkiss::usbInputReserved(input.peek())'), rescue.index('input.read()')) + for role in ('simple_repeater', 'simple_room_server', 'simple_sensor'): + source = (ROOT / 'examples' / role / 'main.cpp').read_text() + signature = ('static void __attribute__((noinline)) serviceCommandInterfaces()' + if role == 'simple_repeater' else 'void loop()') + pump = body(source, signature) + self.assertLess(pump.index('mesh::hostkiss::usbInputReserved(console.peek())'), + pump.index('console.read()')) + + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_host_kiss_usb.py b/test/test_host_kiss_usb.py index fe3717bd8..e91614735 100644 --- a/test/test_host_kiss_usb.py +++ b/test/test_host_kiss_usb.py @@ -113,7 +113,8 @@ static unsigned ascii_calls=0,usb_binary_calls=0,network_calls=0,kiss_resets=0; static std::deque input; struct Port {} terminal_port; struct Usb { - bool enabled=true,passthrough=false; + bool enabled=true,passthrough=false,receiving=false; + bool isReceivingFrame()const{return receiving;} unsigned resets=0,frames=19,enables=0,disables=0; bool isEnabled() const { return enabled; } void enable(){enabled=true;++enables;} @@ -150,7 +151,7 @@ void cancelReaderCommand(Port* p){assert(p==&terminal_port);++reader_cancels;} void discardUsbTerminalOutput(){++text_discards;} void setUsbBinaryStreamOwned(bool value){binary_owned=value;} namespace ota { -struct Source {void resetSessionState(){++mota_resets;}} source; +struct Source {bool pending=false;bool hasPendingResponse()const{return pending;}void resetSessionState(){++mota_resets;}} source; struct OtaContext {static Source& serialFolderSource(){return source;}}; } namespace hostkiss { @@ -179,7 +180,17 @@ static bool poll(){ int main(){ input={'v','e','r','\r'};assert(!poll() && ascii_calls==1 && !serial_cancels); #if MESH_ENABLE_HOST_KISS - terminal_mode=true;usb_mota_mode=true;usb_terminal_discard_line=usb_logging_terminal_mode=true; + // Existing binary ownership cannot be stolen by a FEND payload byte. + usb_serial_interface.receiving=true; input={0xc0}; + assert(!serviceCompanionHostKissUsb() && !kiss_active && input.size()==1); + usb_serial_interface.receiving=false; +#if COMPANION_FEATURE_USB_MOTA_SOURCE + usb_mota_mode=true;assert(!serviceCompanionHostKissUsb() && !kiss_active && input.size()==1); + usb_mota_mode=false;mesh::ota::source.pending=true; + assert(!serviceCompanionHostKissUsb() && !kiss_active && input.size()==1); + mesh::ota::source.pending=false; +#endif + terminal_mode=true;usb_terminal_discard_line=usb_logging_terminal_mode=true; input={0xc0,0x00,0x41};assert(poll()); assert(kiss_active && host_kiss_usb_parked && host_kiss_usb_restore_enabled && !usb_serial_interface.enabled); assert(serial_cancels==1 && route_cancels==1 && forgotten_routes==1); diff --git a/test/test_meshtext_license.py b/test/test_meshtext_license.py index 186a59f36..92fa53bd0 100644 --- a/test/test_meshtext_license.py +++ b/test/test_meshtext_license.py @@ -50,7 +50,7 @@ class MeshtextLicenseTest(unittest.TestCase): build = root / "build/base" build.mkdir(parents=True) elf32(build / "firmware.elf", symbols=symbols) - config = [["env:base", [["build_flags", ["-DCOMPANION_RADIO_FULL=1", *flags]], + config = [["env:base", [["build_flags", ["-DCOMPANION_RADIO_FULL=1", "-DMESH_ENABLE_HOST_KISS=0", *flags]], ["build_unflags", list(config_unflags)]]]] script = r''' set -euo pipefail