mirror of
https://github.com/mikecarper/MeshCore.git
synced 2026-11-04 15:38:24 +00:00
90 lines
4.4 KiB
Python
90 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Run the production advert parser and anonymous repeater handlers under ASan/UBSan.
|
|
|
|
Radio/allocator boundaries are doubles. Path decoding, all three anonymous
|
|
handlers, their receive dispatcher, and the advert implementation are real.
|
|
"""
|
|
|
|
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
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
FIXTURE = ROOT / "test/fixtures/protocol_parser_bounds/test.cpp"
|
|
SANITIZERS = (["-fsanitize=address,undefined", "-fno-sanitize-recover=all",
|
|
"-fno-pie", "-no-pie"] if sys.platform.startswith("linux") else [])
|
|
|
|
|
|
class ProtocolParserBoundsTests(unittest.TestCase):
|
|
def test_production_parsers_with_exact_buffers_and_all_bridge_modes(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")
|
|
repeater = (ROOT / "examples/simple_repeater/MyMesh.cpp").read_text()
|
|
packet = (ROOT / "src/Packet.cpp").read_text()
|
|
identity = (ROOT / "src/Identity.cpp").read_text()
|
|
handlers = "\n".join(extract_braced(repeater, signature) for signature in (
|
|
"static bool hasCompleteAnonReplyPath(",
|
|
"uint8_t MyMesh::handleAnonRegionsReq(",
|
|
"uint8_t MyMesh::handleAnonOwnerReq(",
|
|
"uint8_t MyMesh::handleAnonClockReq(",
|
|
"void MyMesh::onAnonDataRecv(",
|
|
))
|
|
constructors = "namespace mesh {\n" + "\n".join(
|
|
extract_braced(packet, signature) for signature in (
|
|
"Packet::Packet()", "bool Packet::isValidPathLen(",
|
|
"size_t Packet::writePath("))
|
|
constructors += "\n" + extract_braced(identity, "Identity::Identity()") + "\n}\n"
|
|
constants = "\n".join(re.findall(
|
|
r"^#define (?:ANON_REQ_TYPE_(?:REGIONS|OWNER|BASIC)|RESP_SERVER_LOGIN_OK)\s+.+$",
|
|
repeater, re.MULTILINE))
|
|
constants += "\n" + re.search(
|
|
r"static constexpr size_t MAX_ANON_REPLY_LEN =[^;]+;", repeater).group()
|
|
constants += "\n" + re.search(
|
|
r"static constexpr uint32_t LOGIN_PATH_OBSERVATION_TIMEOUT_MS =[^;]+;",
|
|
repeater).group()
|
|
constants += "\n" + re.search(
|
|
r"^\s*#define SERVER_RESPONSE_DELAY\s+.+$", repeater, re.MULTILINE).group()
|
|
constants += "\n" + re.search(
|
|
r"^#define OUT_PATH_UNKNOWN\s+.+$",
|
|
(ROOT / "src/helpers/ContactInfo.h").read_text(), re.MULTILINE).group()
|
|
constants += "\n" + re.search(
|
|
r"^#define REGION_DENY_FLOOD\s+.+$",
|
|
(ROOT / "src/helpers/RegionMap.h").read_text(), re.MULTILINE).group()
|
|
|
|
with tempfile.TemporaryDirectory(prefix="meshcore-parser-bounds-") as directory:
|
|
work = Path(directory)
|
|
(work / "production_handlers.inc").write_text(handlers)
|
|
(work / "production_packet.inc").write_text(constructors)
|
|
(work / "production_constants.inc").write_text(constants)
|
|
variants = (("ordinary", []), ("rs232", ["-DWITH_RS232_BRIDGE=1"]),
|
|
("espnow", ["-DWITH_ESPNOW_BRIDGE=1"]),
|
|
("both", ["-DWITH_RS232_BRIDGE=1", "-DWITH_ESPNOW_BRIDGE=1"]))
|
|
for name, defines in variants:
|
|
with self.subTest(bridge=name):
|
|
binary = work / f"bounds-{name}"
|
|
built = subprocess.run([
|
|
compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
|
|
"-Wno-unused-parameter", "-Wno-reorder", "-Wno-sign-compare",
|
|
*SANITIZERS, *defines, f"-I{work}", f"-I{ROOT / 'src'}",
|
|
f"-I{ROOT / 'test/mocks'}", f"-I{ROOT}",
|
|
str(FIXTURE), str(ROOT / "src/helpers/AdvertDataHelpers.cpp"),
|
|
"-o", str(binary),
|
|
], capture_output=True, text=True, timeout=60)
|
|
self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
|
|
checked = subprocess.run([str(binary)], capture_output=True, text=True, timeout=30)
|
|
self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr)
|
|
self.assertIn("PASS: production protocol parser bounds", checked.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|