rich/MeshCore
1
0 Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Fit padded ACL replies through the Companion frame envelope

Browse Source
This commit is contained in:
mikecarper committed 2026-10-04 14:57:15 -07:00
1 parent 1d96725e73
commit 3a016dd13b
10 files changed
+196 -64
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/helpers/BaseSerialInterface.h
+1 -2
View File
@@ -1,8 +1,7 @@
#pragma once
#include <Arduino.h>
#define MAX_FRAME_SIZE 176 // +4 for transport codes (region scoping)
#include "CompanionFrameLimits.h"
class BaseSerialInterface {
protected:
src/helpers/ClientACLResponse.h
+7 -1
View File
@@ -2,15 +2,21 @@
#include <Packet.h>
#include "DatagramPayloadLimits.h"
#include "CompanionFrameLimits.h"
namespace mesh {
// Match createDatagram() and createPathReturn() admission limits. A flood
// request returns its observed path inside the encrypted response, so those
// bytes must also be reserved before adding legacy seven-byte ACL entries.
static constexpr size_t CLIENT_ACL_DIRECT_REPLY_CAPACITY =
static constexpr size_t CLIENT_ACL_DATAGRAM_REPLY_CAPACITY =
DatagramPayloadLimits::maxPlaintext(
MAX_PACKET_PAYLOAD, CIPHER_MAC_SIZE, CIPHER_BLOCK_SIZE);
static constexpr size_t CLIENT_ACL_HOST_REPLY_CAPACITY =
companionBinaryPlaintextCapacity(CIPHER_BLOCK_SIZE);
static constexpr size_t CLIENT_ACL_DIRECT_REPLY_CAPACITY =
CLIENT_ACL_DATAGRAM_REPLY_CAPACITY < CLIENT_ACL_HOST_REPLY_CAPACITY
? CLIENT_ACL_DATAGRAM_REPLY_CAPACITY : CLIENT_ACL_HOST_REPLY_CAPACITY;
inline size_t clientACLReplyCapacity(bool flood, uint8_t path_len) {
if (!Packet::isValidPathLen(path_len)) return 0;
src/helpers/CompanionFrameLimits.h
+17
View File
@@ -0,0 +1,17 @@
#pragma once
#include <stddef.h>
#define MAX_FRAME_SIZE 176 // +4 for transport codes (region scoping)
namespace mesh {
// Binary responses add an opcode and reserved byte to decrypted radio data.
// A direct datagram exposes its zero padding, so reserve whole cipher blocks
// that fit the host frame rather than counting only unpadded ACL entries.
constexpr size_t companionBinaryPlaintextCapacity(size_t cipher_block_size) {
return cipher_block_size > 0
? ((MAX_FRAME_SIZE - 2) / cipher_block_size) * cipher_block_size : 0;
}
} // namespace mesh
test/fixtures/client_acl_response/test_client_acl_response.cpp
Vendored
+6 -1
View File
@@ -153,6 +153,11 @@ int main() {
PAYLOAD_TYPE_RESPONSE, mesh.reply_data, length)
: mesh.createDatagram(PAYLOAD_TYPE_RESPONSE, mesh.sender.id, secret, mesh.reply_data, length);
assert(packet && packet->payload_len <= MAX_PACKET_PAYLOAD);
const size_t cipher_bytes = packet->payload_len - 2 * PATH_HASH_SIZE - CIPHER_MAC_SIZE;
const size_t path_prefix = flood
? 2 + (path_len & 63) * ((path_len >> 6) + 1) : 0;
assert(cipher_bytes >= path_prefix);
assert(cipher_bytes - path_prefix + 2 <= MAX_FRAME_SIZE);
if (path_len == 0 && !sparse) {
// Decode the actual zero-padded encrypted response body, with the
// reflected tag and flood path envelope removed by Companion.
@@ -174,7 +179,7 @@ int main() {
mesh.sender.permissions = 3;
uint8_t app_query[] = {REQ_TYPE_GET_ACCESS_LIST, 0, 0, 0xA1, 0xB2, 0xC3, 0xD4};
const int app_reply = mesh.handleRequest(&mesh.sender, 51, app_query, sizeof(app_query));
assert(verify_entries(mesh, app_reply, mesh::CLIENT_ACL_DIRECT_REPLY_CAPACITY) == 23);
assert(verify_entries(mesh, app_reply, mesh::CLIENT_ACL_DIRECT_REPLY_CAPACITY) == 22);
assert(mesh.handleRequest(nullptr, 51, query, sizeof(query)) == 0);
for (size_t length : {0, 1, 2}) assert(mesh.handleRequest(&mesh.sender, 51, query, length) == 0);
for (unsigned index : {1, 2}) {
test/fixtures/companion_delayed_reply_delivery/test.cpp
Vendored
+3 -1
View File
@@ -79,6 +79,7 @@ struct BaseChatMesh {
RadioBoundary* _radio = &radio;
Identity self_id;
mesh::Packet packet;
Bytes transmitted_payload;
unsigned allocations = 0, transmissions = 0, connection_starts = 0;
uint32_t txt_send_timeout = 0;
uint32_t direct_timeout = 5000;
@@ -93,7 +94,8 @@ struct BaseChatMesh {
bool onContactPathRecv(ContactInfo&, uint8_t*, uint8_t, uint8_t*, uint8_t,
uint8_t, uint8_t*, uint8_t);
mesh::Packet* allocate() { ++allocations; return &packet; }
mesh::Packet* createDatagram(uint8_t, const Identity&, const uint8_t*, const uint8_t*, size_t) {
mesh::Packet* createDatagram(uint8_t, const Identity&, const uint8_t*, const uint8_t* data, size_t len) {
transmitted_payload.assign(data, data + len);
return allocate();
}
mesh::Packet* createAnonDatagram(uint8_t, const Identity&, const Identity&,
test/fixtures/official_app_compatibility/README.txt
Vendored
+14 -4
View File
@@ -33,13 +33,23 @@ widget state/localization and RNG. The app's schema/tag matching/decoders are
not reimplemented. The deterministic RNG only makes a synthetic request
repeatable; its random value is not the Companion's replay timestamp.
The exact app-generated seven-byte ACL query is passed to the production
repeater receive/ACL handler, with actual packet budget/admission/serialization.
Only the crypto primitive and radio queue are mocked in that native fixture;
zero padding reflects the production cipher block size, not crypto validation.
The exact app-generated CMD50 frame goes through the production Companion
command handler and BaseChat request producer. Its real unique-clock-tagged
11-byte plaintext reaches the production repeater receive/ACL handler, with
actual packet budget/admission/serialization. The server's reflected tag and
full padded plaintext then go through the production Companion response
callback, delayed reply owner and bounded serial transport. Only admitted SENT
and Binary frame bytes reach the stock app frame parser/listener/ACL decoder.
The crypto primitive and radio queue are mocked; zero padding reflects the
production cipher block size, not encryption/authentication validation.
The one-admin direct case asserts 11-byte plaintext, 16-byte padded ciphertext,
20-byte payload and 22-byte zero-path wire size. All 32 decoded cases cover
0/1/22/23/24/25/32/256 entries, direct/flood and zero/maximum return path.
Direct replies are bounded to 22 rows: 23 rows create 176 padded plaintext
bytes, exceeding Companion's 174-byte Binary envelope. An explicit historical
23-row negative uses all 176 bytes and verifies that the real Companion callback
admits no Binary frame. Flood replies include their returned path before cipher
padding; their actual padded extra data also passes through the callback.
Large ACLs retain legacy one-packet truncation, not a complete list guarantee.
Actual Companion and CommonCLI radio replies cover saved automatic/explicit
preambles and an active temporary/secondary profile. A fifth field must invoke
test/fixtures/official_app_compatibility/app_contract.js
Vendored
+3 -2
View File
@@ -149,10 +149,11 @@ async function main() {
let decoded = 0;
for (const reply of input.acl) {
const client = connection(); const promise = client.aQ1(key); await settle();
client.aL9(sent(51)); await settle();
assert.equal(client.writes[0].toString("hex"), builder.writes[0].toString("hex"));
client.aL9(Buffer.from(reply.sent, "hex")); await settle();
const timer = timers[timers.length - 1]; assert.equal(timer.due - now, 7000);
// Valid matching ACL data remains accepted past the old 2400ms deadline.
await advance(3000); client.aL9(binary(51, reply.body)); await settle();
await advance(3000); client.aL9(Buffer.from(reply.frame, "hex")); await settle();
const result = await promise;
assert.equal(result.a.length, reply.entries);
for (let index = 0; index < result.a.length; ++index) {
test/test_companion_delayed_reply_delivery.py
+45 -41
View File
@@ -71,54 +71,58 @@ def production_methods(source, base, packet, dispatcher):
return methods
def production_delayed_reply_inputs(work):
source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text()
base = (ROOT / "src/helpers/BaseChatMesh.cpp").read_text()
packet = (ROOT / "src/Packet.cpp").read_text()
dispatcher = (ROOT / "src/Dispatcher.cpp").read_text()
native = (ROOT / "test/test_serial_mode_switch/test_serial_mode_switch.cpp").read_text()
core = (ROOT / "src/MeshCore.h").read_text()
constants = "\n".join(re.findall(
r"^#define (?:PUSH_CODE_\w+|RESP_CODE_\w+|CMD_SEND_\w+|ERR_CODE_\w+|REQ_TYPE_GET_TELEMETRY_DATA)\s+.+$",
source, re.MULTILINE))
wanted = ("PUB_KEY_SIZE", "MAX_PACKET_PAYLOAD", "MAX_PATH_SIZE", "OUT_PATH_UNKNOWN",
"MSG_SEND_FAILED", "MSG_SEND_SENT_DIRECT", "MSG_SEND_SENT_FLOOD",
"PAYLOAD_TYPE_RESPONSE", "PAYLOAD_TYPE_REQ", "PAYLOAD_TYPE_ANON_REQ",
"PAYLOAD_TYPE_ACK", "REQ_TYPE_GET_STATUS", "REQ_TYPE_GET_TELEMETRY_DATA",
"TELEM_PERM_BASE", "ADV_TYPE_NONE", "ADV_TYPE_ROOM", "RESP_SERVER_LOGIN_OK")
for relative in ("examples/companion_radio/MyMesh.h", "src/MeshCore.h", "src/Packet.h", "src/helpers/BaseChatMesh.h",
"src/helpers/ContactInfo.h", "src/helpers/AdvertDataHelpers.h",
"src/helpers/SensorManager.h"):
text = (ROOT / relative).read_text()
constants += "\n" + "\n".join(re.findall(
r"^#define (?:" + "|".join(wanted) + r")\s+.+$", text, re.MULTILINE))
constants += "\n#define EXPECTED_ACK_TABLE_SIZE 2\n"
clock = "\n".join(extract_braced(core, signature) for signature in (
"uint32_t getCurrentTimeUnique()", "void resetUniqueTime(uint32_t time)"))
# printf is an Arduino Print operation absent from the lean shared
# host mock. Add only this I/O boundary, leaving transport logic real.
stream = (ROOT / "test/mocks/Stream.h").read_text()
stream = "#include <cstdarg>\n#include <cstdio>\n" + stream.replace(
"public:", "public:\n size_t printf(const char* format, ...) {\n"
" char text[256]; va_list args; va_start(args, format);\n"
" int n=vsnprintf(text,sizeof(text),format,args); va_end(args);\n"
" return n>0 ? write((const uint8_t*)text, (size_t)n<sizeof(text) ? (size_t)n : sizeof(text)-1) : 0;\n"
" }", 1)
(work / "production_constants.inc").write_text(constants, encoding="ascii")
(work / "production_clock.inc").write_text(clock, encoding="ascii")
(work / "production_stream.inc").write_text(
extract_braced(native, "class BufferStream") + ";\n", encoding="ascii")
(work / "production_replies.inc").write_text(
production_methods(source, base, packet, dispatcher), encoding="ascii")
(work / "Stream.h").write_text(stream, encoding="ascii")
(work / "Arduino.h").write_text(
(ROOT / "test/mocks/Arduino.h").read_text(), encoding="ascii")
class CompanionDelayedReplyDeliveryTests(unittest.TestCase):
def test_actual_producers_routes_deadlines_and_required_admission(self):
compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
if compiler is None:
self.skipTest("a host C++17 compiler is required")
source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text()
base = (ROOT / "src/helpers/BaseChatMesh.cpp").read_text()
packet = (ROOT / "src/Packet.cpp").read_text()
dispatcher = (ROOT / "src/Dispatcher.cpp").read_text()
native = (ROOT / "test/test_serial_mode_switch/test_serial_mode_switch.cpp").read_text()
core = (ROOT / "src/MeshCore.h").read_text()
constants = "\n".join(re.findall(
r"^#define (?:PUSH_CODE_\w+|RESP_CODE_\w+|CMD_SEND_\w+|ERR_CODE_\w+|REQ_TYPE_GET_TELEMETRY_DATA)\s+.+$",
source, re.MULTILINE))
wanted = ("PUB_KEY_SIZE", "MAX_PACKET_PAYLOAD", "MAX_PATH_SIZE", "OUT_PATH_UNKNOWN",
"MSG_SEND_FAILED", "MSG_SEND_SENT_DIRECT", "MSG_SEND_SENT_FLOOD",
"PAYLOAD_TYPE_RESPONSE", "PAYLOAD_TYPE_REQ", "PAYLOAD_TYPE_ANON_REQ",
"PAYLOAD_TYPE_ACK", "REQ_TYPE_GET_STATUS", "REQ_TYPE_GET_TELEMETRY_DATA",
"TELEM_PERM_BASE", "ADV_TYPE_NONE", "ADV_TYPE_ROOM", "RESP_SERVER_LOGIN_OK")
for relative in ("examples/companion_radio/MyMesh.h", "src/MeshCore.h", "src/Packet.h", "src/helpers/BaseChatMesh.h",
"src/helpers/ContactInfo.h", "src/helpers/AdvertDataHelpers.h",
"src/helpers/SensorManager.h"):
text = (ROOT / relative).read_text()
constants += "\n" + "\n".join(re.findall(
r"^#define (?:" + "|".join(wanted) + r")\s+.+$", text, re.MULTILINE))
constants += "\n#define EXPECTED_ACK_TABLE_SIZE 2\n"
clock = "\n".join(extract_braced(core, signature) for signature in (
"uint32_t getCurrentTimeUnique()", "void resetUniqueTime(uint32_t time)"))
# printf is an Arduino Print operation absent from the lean shared
# host mock. Add only this I/O boundary, leaving transport logic real.
stream = (ROOT / "test/mocks/Stream.h").read_text()
stream = "#include <cstdarg>\n#include <cstdio>\n" + stream.replace(
"public:", "public:\n size_t printf(const char* format, ...) {\n"
" char text[256]; va_list args; va_start(args, format);\n"
" int n=vsnprintf(text,sizeof(text),format,args); va_end(args);\n"
" return n>0 ? write((const uint8_t*)text, (size_t)n<sizeof(text) ? (size_t)n : sizeof(text)-1) : 0;\n"
" }", 1)
with tempfile.TemporaryDirectory(prefix="meshcore-delayed-replies-") as directory:
work = Path(directory)
(work / "production_constants.inc").write_text(constants, encoding="ascii")
(work / "production_clock.inc").write_text(clock, encoding="ascii")
(work / "production_stream.inc").write_text(
extract_braced(native, "class BufferStream") + ";\n", encoding="ascii")
(work / "production_replies.inc").write_text(
production_methods(source, base, packet, dispatcher), encoding="ascii")
(work / "Stream.h").write_text(stream, encoding="ascii")
(work / "Arduino.h").write_text(
(ROOT / "test/mocks/Arduino.h").read_text(), encoding="ascii")
production_delayed_reply_inputs(work)
for terminal in (0, 1):
with self.subTest(terminal=terminal):
binary = work / f"delayed-{terminal}.exe"
test/test_companion_response_bounds.py
+1 -1
View File
@@ -148,7 +148,7 @@ class CompanionResponseBoundsTest(unittest.TestCase):
def test_response_envelopes(self):
source = (ROOT / 'examples/companion_radio/MyMesh.cpp').read_text(encoding='utf-8')
codes = '\n'.join(re.findall(r'^#define PUSH_CODE_\w+\s+0x[0-9A-Fa-f]+', source, re.M))
interface = (ROOT / 'src/helpers/BaseSerialInterface.h').read_text(encoding='utf-8')
interface = (ROOT / 'src/helpers/CompanionFrameLimits.h').read_text(encoding='utf-8')
frame_limit = re.search(r'^#define MAX_FRAME_SIZE\s+\d+', interface, re.M).group(0)
methods = '\n'.join(extract_braced(source, signature) for signature in (
'void MyMesh::onContactResponse(', 'void MyMesh::onControlDataRecv(',
test/test_official_app_compatibility.py
+99 -11
View File
@@ -20,6 +20,7 @@ import urllib.request
from test_client_acl_response import production_acl_methods
from test_companion_primary_radio_persistence import production_primary_radio_harness
from test_companion_delayed_reply_delivery import production_delayed_reply_inputs
from test_replay_reset_integration import extract_braced
ROOT = Path(__file__).resolve().parents[1]
@@ -57,7 +58,7 @@ def checked(command, timeout=60):
return result.stdout
def acl_outputs(work, compiler, app_request):
def acl_outputs(work, compiler, request_plaintext):
source = production_acl_methods()
packet = (ROOT / "src/Packet.cpp").read_text()
source += "namespace mesh {\n" + "\n".join(extract_braced(packet, signature) for signature in (
@@ -66,19 +67,18 @@ def acl_outputs(work, compiler, app_request):
(work / "production.inc").write_text(source, encoding="ascii")
fixture = (ROOT / "test/fixtures/client_acl_response/test_client_acl_response.cpp").read_text()
fixture = fixture[:fixture.index("int main()")]
request = ",".join(str(byte) for byte in bytes.fromhex(app_request))
request = ",".join(str(byte) for byte in bytes.fromhex(request_plaintext))
fixture += r'''
int main() {
uint8_t app_query[] = {APP_QUERY};
static_assert(sizeof(app_query) == 7, "Exact stock app ACL request schema");
uint8_t plaintext[] = {APP_QUERY};
static_assert(sizeof(plaintext) == 11, "Actual Companion tag plus stock seven-byte query");
uint32_t tag; memcpy(&tag, plaintext, 4);
for (bool flood : {false, true}) for (uint8_t path_len : {0, 0x60})
for (unsigned clients : {0, 1, 22, 23, 24, 25, 32, 256}) {
MyMesh target; fill(target.acl, clients);
target.sender.last_timestamp = 50;
target.sender.last_timestamp = tag - 1;
mesh::Packet request; request.header = flood ? ROUTE_TYPE_FLOOD : ROUTE_TYPE_DIRECT;
request.path_len = path_len;
uint8_t plaintext[11]; uint32_t tag = 51;
memcpy(plaintext, &tag, 4); memcpy(plaintext + 4, app_query, sizeof(app_query));
uint8_t secret[PUB_KEY_SIZE] = {};
auto* transmitted = target.createDatagram(PAYLOAD_TYPE_REQ, target.sender.id,
secret, plaintext, sizeof(plaintext));
@@ -92,7 +92,7 @@ int main() {
const unsigned capacity = mesh::clientACLReplyCapacity(flood, path_len);
const unsigned entries = std::min(clients, (capacity - 4) / 7);
const unsigned route_bytes = (path_len & 63) * ((path_len >> 6) + 1);
const size_t body_offset = 2 * PATH_HASH_SIZE + CIPHER_MAC_SIZE + 4
const size_t body_offset = 2 * PATH_HASH_SIZE + CIPHER_MAC_SIZE
+ (flood ? 2 + route_bytes : 0);
if (!flood && clients == 1) {
assert(sizeof(plaintext) == 11 && reply->payload_len == 20 && reply->getRawLength() == 22);
@@ -112,13 +112,99 @@ int main() {
records = []
for line in output.splitlines():
if line.startswith("LEGACY:"):
_, entries, body = line.split(":")
records.append({"entries": int(entries), "body": body})
_, entries, plaintext = line.split(":")
records.append({"entries": int(entries), "plaintext": plaintext})
if len(records) != 32:
raise AssertionError(output)
return records
def companion_pipeline(work, compiler, command):
production_delayed_reply_inputs(work)
fixture = (ROOT / "test/fixtures/companion_delayed_reply_delivery/test.cpp").read_text()
fixture = fixture[:fixture.index("int main()")]
fixture += r'''
static Bytes fromHex(const char* text) {
Bytes bytes;
for (size_t at = 0; text[at]; at += 2) {
unsigned byte; assert(text[at + 1]); assert(sscanf(text + at, "%2x", &byte) == 1);
bytes.push_back(uint8_t(byte));
}
return bytes;
}
static void printHex(const char* prefix, const Bytes& bytes) {
printf("%s:", prefix); for (uint8_t byte : bytes) printf("%02x", byte); printf("\n");
}
int main(int argc, char** argv) {
assert(argc == 2 || argc == 4);
const Bytes command = fromHex(argv[1]); assert(command.size() == 40);
Fixture f; f.mesh.direct_timeout = 2000;
memcpy(f.mesh.recipient.id.pub_key, command.data() + 1, 32);
memcpy(f.mesh.cmd_frame, command.data(), command.size());
f.mesh.handleRequestFrame(command.size());
const auto& payload = f.mesh.transmitted_payload;
assert(payload.size() == 11 && !memcmp(payload.data() + 4, command.data() + 33, 7));
uint32_t tag; memcpy(&tag, payload.data(), 4);
assert(tag == slot(f, Tracker::Binary).tag && !slot(f, Tracker::Binary).sent_pending);
if (argc == 2) { printHex("REQUEST", payload); return 0; }
Bytes radio = fromHex(argv[2]);
if (atoi(argv[3]) == 0) {
// The old 23-row direct reply was 165 bytes before zero padding. The
// decrypted 176 bytes exceed Companion's 174-byte callback envelope.
radio.assign(176, 0); memcpy(radio.data(), &tag, 4);
for (unsigned entry = 0; entry < 23; ++entry) {
for (unsigned byte = 0; byte < 6; ++byte) radio[4 + entry * 7 + byte] = entry + byte + 1;
radio[4 + entry * 7 + 6] = 3;
}
}
assert(radio.size() <= 255 && !memcmp(radio.data(), &tag, 4));
f.otherRequest(); g_mock_millis += 3000;
f.mesh.onContactResponse(f.mesh.recipient, radio.data(), uint8_t(radio.size())); f.drain();
const bool admitted = atoi(argv[3]) != 0;
assert(count(f.wire(), PUSH_CODE_BINARY_RESPONSE) == (admitted ? 1U : 0U));
assert(count(f.wire(), RESP_CODE_SENT) == 1 && f.other_stream.output.empty());
if (admitted) {
assertSentBeforeFinal(f.wire(), Tracker::Binary, tag);
assert(frameWith(f.wire(), PUSH_CODE_BINARY_RESPONSE).size() <= MAX_FRAME_SIZE);
} else assert(radio.size() == 176);
for (const auto& frame : frames(f.wire())) printHex("FRAME", frame);
}
'''
(work / "companion.cpp").write_text(fixture, encoding="ascii")
binary = work / "companion.exe"
checked([compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
"-Wno-unused-parameter", "-Wno-unused-function", "-Wno-sign-compare", "-Wno-class-memaccess",
*SANITIZERS, "-DCOMPANION_FEATURE_TEXT_TERMINAL=0", "-DMESH_ENABLE_ONE_KEY_DM=0",
f"-I{work}", f"-I{ROOT / 'test/mocks'}",
f"-I{ROOT / 'test/fixtures/serial_wifi_sessions/mocks'}", f"-I{ROOT / 'src'}",
str(work / "companion.cpp"), *[str(ROOT / "src/helpers" / name) for name in (
"CompanionDelayedReplies.cpp", "ArduinoSerialInterface.cpp",
"wifi/SerialWifiInterface.cpp", "TxtDataHelpers.cpp")], "-o", str(binary)])
output = checked([str(binary), command], timeout=20)
plaintext = output.strip().removeprefix("REQUEST:")
if len(bytes.fromhex(plaintext)) != 11:
raise AssertionError(output)
def deliver(records):
forwarded = []
for record in records:
output = checked([str(binary), command, record["plaintext"], "1"], timeout=20)
frames = [line.removeprefix("FRAME:") for line in output.splitlines() if line.startswith("FRAME:")]
if len(frames) != 2:
raise AssertionError(output)
forwarded.append({"entries": record["entries"], "sent": frames[0], "frame": frames[1],
"body": record["plaintext"][8:]})
# The unmodified production callback must reject the historical padded
# 23-entry reply. The app therefore cannot falsely decode this fixture.
output = checked([str(binary), command, plaintext, "0"], timeout=20)
rejected = [line.removeprefix("FRAME:") for line in output.splitlines() if line.startswith("FRAME:")]
if len(rejected) != 1 or bytes.fromhex(rejected[0])[0] != 6:
raise AssertionError(output)
return forwarded
return plaintext, deliver
def radio_outputs(work, compiler):
harness = production_primary_radio_harness()
# Execute the same production Companion dispatch, shared preferences parser
@@ -177,8 +263,10 @@ class OfficialAppCompatibilityTests(unittest.TestCase):
config = work / "input.json"
config.write_text(json.dumps({"bundle": str(bundle), "mode": "build"}), encoding="ascii")
generated = json.loads(checked([node, str(JS), str(config)], timeout=20))
plaintext, deliver = companion_pipeline(work, compiler, generated["command"])
acl = deliver(acl_outputs(work, compiler, plaintext))
config.write_text(json.dumps({"bundle": str(bundle), "mode": "check",
"acl": acl_outputs(work, compiler, generated["request"]),
"acl": acl,
"radio": radio_outputs(work, compiler)}), encoding="ascii")
result = json.loads(checked([node, str(JS), str(config)], timeout=20))
self.assertEqual(result["acl_cases"], 32)