#!/usr/bin/env python3 """Reject high-risk ESP stack allocations in protocol hot paths.""" from __future__ import annotations from dataclasses import dataclass from pathlib import Path import os import re import subprocess import sys REPO_ROOT = Path(__file__).resolve().parent.parent SOURCE_SUFFIXES = {".c", ".cc", ".cpp", ".cxx", ".h", ".hpp", ".hh"} IMPLEMENTATION_SUFFIXES = {".c", ".cc", ".cpp", ".cxx"} HOT_PATH_PREFIXES = ( "platform/esp/arduino_common/src/chat/infra/meshtastic/", "modules/core_phone/src/meshtastic/", ) HOT_PATH_FILES = { "platform/esp/arduino_common/src/ble/meshtastic_ble.cpp", } HOT_HEADER_PREFIXES = ( "platform/esp/arduino_common/include/platform/esp/arduino_common/chat/infra/meshtastic/", "modules/core_phone/include/phone/meshtastic/", ) HOT_HEADER_FILES = { "platform/esp/arduino_common/include/ble/meshtastic_ble.h", } APP_CONFIG_PATHS = { "platform/esp/arduino_common/src/app_config_store.cpp", "platform/esp/arduino_common/src/app_context.cpp", } HIGH_RISK_PROTOCOL_TYPES = ( "MeshtasticBleFrame", "phone::meshtastic::MeshtasticBleFrame", "meshtastic_FromRadio", "meshtastic_ToRadio", "meshtastic_MeshPacket", "meshtastic_Data", "meshtastic_AdminMessage", "meshtastic_MqttClientProxyMessage", "meshtastic_ServiceEnvelope", ) HIGH_RISK_CONFIG_TYPES = ( "AppConfig", "app::AppConfig", "MeshConfig", "chat::MeshConfig", ) RAW_ARRAY_STACK_LIMIT = 192 LARGE_ARRAY_SIZE_TOKENS = ( "MeshtasticBleFrame::kMaxPayload", "meshtastic_FromRadio_size", "meshtastic_ToRadio_size", "kMaxEncoded", "MAX_PACKET_SIZE", "TM_C6_MAX_PAYLOAD", ) @dataclass(frozen=True) class Violation: path: Path line_number: int rule: str line: str detail: str def normalize(path: Path) -> str: return path.relative_to(REPO_ROOT).as_posix() def is_under(relative: str, prefixes: tuple[str, ...]) -> bool: return any(relative.startswith(prefix) for prefix in prefixes) def git_tracked_files() -> list[Path]: try: result = subprocess.run( ["git", "ls-files"], cwd=REPO_ROOT, check=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) except (OSError, subprocess.CalledProcessError): return [] return [REPO_ROOT / line for line in result.stdout.splitlines() if line] def walk_source_files() -> list[Path]: tracked = git_tracked_files() if tracked: return [path for path in tracked if path.suffix.lower() in SOURCE_SUFFIXES] source_files: list[Path] = [] excluded_dirs = {".git", ".pio", ".tmp", "build", "dist", "third_party"} for root, dir_names, file_names in os.walk(REPO_ROOT): dir_names[:] = [name for name in dir_names if name not in excluded_dirs] root_path = Path(root) for file_name in file_names: path = root_path / file_name if path.suffix.lower() in SOURCE_SUFFIXES: source_files.append(path) return source_files def type_decl_pattern(type_name: str) -> re.Pattern[str]: escaped = re.escape(type_name) return re.compile( rf"^\s*(?!static\b)(?:(?:const|volatile)\s+)*{escaped}\s+" rf"([A-Za-z_][A-Za-z0-9_]*)\s*(?:[={{;(])" ) PROTOCOL_TYPE_PATTERNS = tuple( (type_name, type_decl_pattern(type_name)) for type_name in HIGH_RISK_PROTOCOL_TYPES ) CONFIG_TYPE_PATTERNS = tuple( (type_name, type_decl_pattern(type_name)) for type_name in HIGH_RISK_CONFIG_TYPES ) RAW_ARRAY_PATTERN = re.compile( r"^\s*(?!static\b)(?:(?:const|volatile)\s+)*" r"(?:uint8_t|std::uint8_t|char|unsigned\s+char|std::byte)\s+" r"([A-Za-z_][A-Za-z0-9_]*)\s*\[\s*([A-Za-z0-9_:]+)\s*\]" ) STD_ARRAY_PATTERN = re.compile( r"^\s*(?!static\b)(?:(?:const|volatile)\s+)*" r"std::array\s*<\s*(?:uint8_t|std::uint8_t|char|unsigned\s+char|std::byte)\s*,\s*" r"([A-Za-z0-9_:]+)\s*>\s+([A-Za-z_][A-Za-z0-9_]*)" ) STD_DEQUE_PATTERN = re.compile(r"\bstd::deque\s*<") def is_large_array_size(size_token: str) -> bool: if size_token.isdigit(): return int(size_token) >= RAW_ARRAY_STACK_LIMIT return size_token in LARGE_ARRAY_SIZE_TOKENS def collect_violations() -> list[Violation]: violations: list[Violation] = [] for path in walk_source_files(): relative = normalize(path) suffix = path.suffix.lower() in_hot_impl = suffix in IMPLEMENTATION_SUFFIXES and ( relative in HOT_PATH_FILES or is_under(relative, HOT_PATH_PREFIXES) ) in_hot_header = relative in HOT_HEADER_FILES or is_under(relative, HOT_HEADER_PREFIXES) in_app_config = relative in APP_CONFIG_PATHS if not in_hot_impl and not in_hot_header and not in_app_config: continue try: lines = path.read_text(encoding="utf-8", errors="ignore").splitlines() except OSError as exc: violations.append(Violation(path, 0, "read-error", "", str(exc))) continue for line_number, line in enumerate(lines, start=1): stripped = line.strip() if not stripped or stripped.startswith("//"): continue if in_hot_header and STD_DEQUE_PATTERN.search(line): violations.append( Violation( path, line_number, "esp-hot-path-std-deque", stripped, "Use a fixed-depth ring buffer with an explicit drop policy.", ) ) if in_hot_impl: for type_name, pattern in PROTOCOL_TYPE_PATTERNS: if pattern.search(line): violations.append( Violation( path, line_number, "esp-hot-path-large-local", stripped, f"Move local {type_name} storage to member scratch/static storage or fill an existing slot.", ) ) raw_array = RAW_ARRAY_PATTERN.search(line) if raw_array and is_large_array_size(raw_array.group(2)): violations.append( Violation( path, line_number, "esp-hot-path-large-array", stripped, "Move large raw byte arrays off the task stack.", ) ) std_array = STD_ARRAY_PATTERN.search(line) if std_array and is_large_array_size(std_array.group(1)): violations.append( Violation( path, line_number, "esp-hot-path-large-std-array", stripped, "Move large std::array byte buffers off the task stack.", ) ) if in_app_config: for type_name, pattern in CONFIG_TYPE_PATTERNS: if pattern.search(line): violations.append( Violation( path, line_number, "esp-app-config-large-local", stripped, f"Do not place {type_name} on app/config task stacks; use existing context storage or scratch.", ) ) return violations def main() -> int: violations = collect_violations() if not violations: print("ESP stack hygiene check passed.") return 0 print("ESP stack hygiene check failed:") for violation in violations: relative = normalize(violation.path) location = f"{relative}:{violation.line_number}" if violation.line_number else relative print(f"- [{violation.rule}] {location}") if violation.line: print(f" {violation.line}") print(f" {violation.detail}") return 1 if __name__ == "__main__": sys.exit(main())