#!/usr/bin/env python3 """Execute Room/Sensor ACL handlers and encrypted packet assembly with bounds.""" from pathlib import Path import os import re import shutil import subprocess import sys import tempfile import unittest from test_replay_reset_integration import extract_braced from test_client_acl_response import parse_acl ROOT = Path(__file__).resolve().parents[1] FIXTURE = ROOT / "test/fixtures/client_acl_infrastructure/test_client_acl_infrastructure.cpp" def room_mail_crypto_arguments(role): if role == "sensor": return [] candidates = [Path(os.environ["MESHCORE_CRYPTO_DIR"])] if os.environ.get("MESHCORE_CRYPTO_DIR") else [] candidates += sorted((ROOT / ".pio/libdeps").glob("*/Crypto")) crypto = next((path for path in candidates if (path / "SHA256.cpp").is_file()), None) if crypto is None: raise RuntimeError("Install rweather/Crypto or set MESHCORE_CRYPTO_DIR (no hash mock fallback)") return ["-DHOST_BUILD", "-I" + str(crypto), str(crypto / "SHA256.cpp"), str(crypto / "Hash.cpp"), str(crypto / "Crypto.cpp")] def production_methods(role, transform=None): sensor = role == "sensor" path = ROOT / ("examples/simple_sensor/SensorMesh.cpp" if sensor else "examples/simple_room_server/MyMesh.cpp") source = path.read_text() if transform: source = transform(source) name = "SensorMesh" if sensor else "MyMesh" handler = extract_braced(source, ("uint8_t " if sensor else "int ") + name + "::handleRequest(") begin = handler.index(" if (req_type == REQ_TYPE_GET_TELEMETRY_DATA" if sensor else " if (payload[0] == REQ_TYPE_GET_STATUS)") prefix = handler[:begin] # The new board protocol has its own admission/storage regression suite. # Keep the complete mailbox prefix: its role/send gates also protect this # authenticated request dispatcher, and use the actual protocol and store. if not sensor and "if (payload[0] == mesh::ROOM_BOARD_REQUEST_SUBTYPE)" in prefix: board = extract_braced(prefix, "if (payload[0] == mesh::ROOM_BOARD_REQUEST_SUBTYPE)") prefix = prefix.replace(board, "", 1) acl = extract_braced(handler, "if (req_type == REQ_TYPE_GET_ACCESS_LIST" if sensor else "if (payload[0] == REQ_TYPE_GET_ACCESS_LIST") generated = prefix + (" (void)perms;\n" if sensor else "") + acl + "\nreturn 0;\n}\n" signature = "if (type == PAYLOAD_TYPE_REQ) { // request" if sensor else "if (type == PAYLOAD_TYPE_REQ && len >= 5)" receive = extract_braced(source, signature) generated += "void " + name + "::receive(mesh::Packet* packet, uint8_t* data, size_t len) {\n" generated += "ClientInfo* " + ("from" if sensor else "client") + " = &sender;\n" generated += "const uint8_t type = PAYLOAD_TYPE_REQ; uint8_t secret[PUB_KEY_SIZE] = {};\n" + receive + "\n}\n" if not sensor: generated += extract_braced(source, "bool MyMesh::sendClientReply(") + "\n" generated += extract_braced(source, "bool MyMesh::saveFilter(") + "\n" telemetry = extract_braced(handler, "if (payload[0] == REQ_TYPE_GET_TELEMETRY_DATA)") telemetry_prefix = telemetry[:telemetry.index(" telemetry.reset();")] generated += "int MyMesh::telemetryPrefix(uint8_t* payload, size_t payload_len) {\n" generated += "ClientInfo* sender = &this->sender; uint32_t sender_timestamp = 51;\n" generated += "size_t reply_capacity = mesh::CLIENT_ACL_DIRECT_REPLY_CAPACITY;\n" generated += prefix[prefix.index("{") + 1:] + telemetry_prefix + "return perm_mask;\n}\nreturn 0;\n}\n" if sensor: generated += extract_braced(source, "bool SensorMesh::sendFloodScoped(") + "\n" generated += extract_braced(source, "bool SensorMesh::sendFloodReply(") + "\n" mesh_source = (ROOT / "src/Mesh.cpp").read_text() generated += "namespace mesh {\n#define MAX_COMBINED_PATH (MAX_PACKET_PAYLOAD - 2 - CIPHER_BLOCK_SIZE)\n" for signature in ("Packet* Mesh::createPathReturn(const Identity& dest,", "Packet* Mesh::createPathReturn(const uint8_t* dest_hash,", "Packet* Mesh::createDatagram(", "bool Mesh::sendFlood(Packet* packet, uint32_t", "bool Mesh::sendFlood(Packet* packet, uint16_t*", "bool Mesh::sendDirect("): generated += extract_braced(mesh_source, signature) + "\n" packet_source = (ROOT / "src/Packet.cpp").read_text() for signature in ("Packet::Packet()", "bool Packet::isValidPathLen(", "uint8_t Packet::copyPath(", "size_t Packet::writePath("): generated += extract_braced(packet_source, signature) + "\n" return generated + "}\n" class ClientAclInfrastructureTest(unittest.TestCase): def execute(self, role, transform=None): compiler = shutil.which("g++") or shutil.which("clang++") self.assertIsNotNone(compiler, "host C++17 compiler required") with tempfile.TemporaryDirectory(prefix="infrastructure-acl-") as directory: work = Path(directory) (work / "production.inc").write_text(production_methods(role, transform), encoding="ascii") binary = work / "acl.exe" command = [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", "-Wno-unused-parameter", "-I" + str(work), "-I" + str(ROOT / "test/fixtures/room_history_store"), "-I" + str(ROOT / "src"), str(FIXTURE), *room_mail_crypto_arguments(role), "-o", str(binary)] if role == "sensor": command.insert(1, "-DTEST_SENSOR=1") if sys.platform.startswith("linux"): command[1:1] = ["-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie"] built = subprocess.run(command, capture_output=True, text=True, timeout=60) self.assertEqual(built.returncode, 0, built.stdout + built.stderr) run = subprocess.run([str(binary)], capture_output=True, text=True, timeout=30) return run def check_role(self, role): run = self.execute(role) self.assertEqual(run.returncode, 0, run.stdout + run.stderr) self.assertIn("Room/Sensor ACL route, input, and permission checks passed", run.stdout) if role == "room": self.assertIn("Room mailbox guest, reader, writer and recipient-ban boundaries passed", run.stdout) decoded = 0 for line in run.stdout.splitlines(): if not line.startswith("LEGACY:"): continue _, expected, body = line.split(":") entries = parse_acl(bytes.fromhex(body)) self.assertEqual(len(entries), int(expected)) self.assertTrue(all(entry["perm"] == 3 for entry in entries)) decoded += 1 self.assertGreaterEqual(decoded, 10) def test_room_actual_route_capacity_input_and_admin_filter(self): self.check_role("room") def test_sensor_actual_route_capacity_input_and_true_admin_permission(self): self.check_role("sensor") def test_old_acl_loop_is_rejected_for_both_roles(self): def old(source): self.assertIn("ofs + 7 <= reply_capacity", source) return source.replace("ofs + 7 <= reply_capacity", "ofs + 7 <= sizeof(reply_data) - 4", 1) for role in ("room", "sensor"): with self.subTest(role=role): self.assertNotEqual(self.execute(role, old).returncode, 0) def test_old_room_reserved_byte_overread_is_rejected(self): def old(source): self.assertIn(" if (payload_len < 3) return 0;", source) return source.replace(" if (payload_len < 3) return 0;", "", 1) self.assertNotEqual(self.execute("room", old).returncode, 0) def test_room_mailbox_inactive_guest_cannot_bypass_session_gate(self): def old(source): handler = extract_braced(source, "int MyMesh::handleRequest(") mailbox = extract_braced(handler, "if (payload[0] == mesh::ROOM_MAIL_REQUEST_SUBTYPE)") guard = "role == PERM_ACL_GUEST && sender->last_activity != 0" self.assertIn(guard, mailbox) return source.replace(mailbox, mailbox.replace(guard, "role == PERM_ACL_GUEST"), 1) self.assertNotEqual(self.execute("room", old).returncode, 0) def test_room_mailbox_reader_cannot_bypass_send_permission(self): def old(source): handler = extract_braced(source, "int MyMesh::handleRequest(") mailbox = extract_braced(handler, "if (payload[0] == mesh::ROOM_MAIL_REQUEST_SUBTYPE)") guard = "role == PERM_ACL_READ_WRITE || role == PERM_ACL_ADMIN" self.assertIn(guard, mailbox) return source.replace(mailbox, mailbox.replace(guard, "true"), 1) self.assertNotEqual(self.execute("room", old).returncode, 0) def test_old_sensor_truncated_packet_overread_is_rejected(self): def old(source): self.assertIn(" if (len < 5) return;", source) return source.replace(" if (len < 5) return;", "", 1) self.assertNotEqual(self.execute("sensor", old).returncode, 0) def test_old_sensor_effective_permissions_break_genuine_admin(self): def old(source): self.assertIn(" && from->isAdmin())", source) return source.replace(" && from->isAdmin())", " && (perms & PERM_ACL_ROLE_MASK) == PERM_ACL_ADMIN)", 1) self.assertNotEqual(self.execute("sensor", old).returncode, 0) def test_old_force_flood_route_selection_is_rejected(self): for role, client in (("room", "client"), ("sensor", "from")): def old(source): guard = "mesh::Packet::isValidPathLen(" + client + "->out_path_len)" signature = ("bool MyMesh::sendClientReply(" if role == "room" else "if (type == PAYLOAD_TYPE_REQ) { // request") receive = extract_braced(source, signature) self.assertIn(guard, receive) altered = receive.replace(guard, "(" + client + "->out_path_len != OUT_PATH_UNKNOWN)") return source.replace(receive, altered, 1) with self.subTest(role=role): self.assertNotEqual(self.execute(role, old).returncode, 0) def test_sensor_early_replay_commit_prevents_retry_after_rejection(self): def old(source): commit = """ if (reply_queued) { from->last_timestamp = timestamp; from->last_activity = getRTCClock()->getCurrentTime(); }""" self.assertIn(commit, source) return source.replace(commit, " (void)reply_queued;", 1).replace( " bool reply_queued = false;", " from->last_timestamp = timestamp;\n" " from->last_activity = getRTCClock()->getCurrentTime();\n" " bool reply_queued = false;", 1) self.assertNotEqual(self.execute("sensor", old).returncode, 0) def test_sensor_scope_wrappers_must_propagate_failed_queue_admission(self): def old(source): changed, count = re.subn(r"return (sendFlood(?:Scoped)?\([^;]+\));", r"\1; return true;", source) self.assertGreaterEqual(count, 3) return changed self.assertNotEqual(self.execute("sensor", old).returncode, 0) def test_default_capacity_declarations_and_ci_are_wired(self): for path in ("examples/simple_room_server/MyMesh.h", "examples/simple_sensor/SensorMesh.h"): self.assertIn("size_t reply_capacity = MAX_PACKET_PAYLOAD - CIPHER_MAC_SIZE - (CIPHER_BLOCK_SIZE - 1)", (ROOT / path).read_text()) self.assertIn(" python3 -B test/test_client_acl_infrastructure.py -v\n", (ROOT / ".github/workflows/run-unit-tests.yml").read_text()) if __name__ == "__main__": unittest.main()