diff --git a/.gitignore b/.gitignore index eca6fc7..b5abd90 100644 --- a/.gitignore +++ b/.gitignore @@ -78,4 +78,9 @@ tools/pylink-square-mcp/.rtt_state/rtt_read_state.json # Ignore the J-Link Remote Server logs spawned by the MCP server tools/pylink-square-mcp/jlink-remote-server.log -tools/pylink-square-mcp/JLinkRemoteServer.log \ No newline at end of file +tools/pylink-square-mcp/JLinkRemoteServer.log + +# Ignore local PLAN and TODO in favor of Tickets on Github/Gitlab +TODO.md +PLAN.md +HANDOFF.md diff --git a/CMakePresets.json b/CMakePresets.json index e09754e..a74357c 100644 --- a/CMakePresets.json +++ b/CMakePresets.json @@ -49,7 +49,6 @@ "DISTCC_VERBOSE": "0", "DISTCC_FALLBACK": "1", "DISTCC_SKIP_LOCAL_RETRY": "0", - "DISTCC_DIR": "/tmp/distcc", "PATH": "/Applications/ArmGNUToolchain/13.2.rel1/arm-none-eabi/bin:$penv{PATH}" } }, diff --git a/HANDOFF.md b/HANDOFF.md deleted file mode 100644 index 0164652..0000000 --- a/HANDOFF.md +++ /dev/null @@ -1,128 +0,0 @@ -# HANDOFF: Continue implementing Cyphal Services on nucleo-cyphal - -Status: **Discovery phase only. No code written.** Prepared 2026-08-13. - -## What the human asked for - -Implement, next, on the `nucleo-cyphal` application: -1. `uavcan.node.ExecuteCommand` (fixed port-ID **435**) — server side. -2. `uavcan.register.Access` (**384**) / `uavcan.register.List` (**385**) — server side. - -## Current state of nucleo-cyphal (all merged into `develop`) - -The app lives in `applications/nucleo-cyphal/`. Already implemented: -- `uavcan.node.Heartbeat` (7509) publication — every 1 s. -- `uavcan.diagnostic.Record` (8184) publication — every 1 s. -- `uavcan.node.GetInfo` (430) **server** — issue #44. -- `uavcan.node.GetInfo` (430) **client scanner** — issue #45 (`GetInfoScanner` class, host-tested). - -Key files: -- `applications/nucleo-cyphal/include/CyphalApp.hpp` — app state: `service_dispatcher_` (`UdpardRxRPCDispatcher`), `get_info_service_port_`, `get_info_response_port_` (`UdpardRxRPCPort`), `service_group_address_`. -- `applications/nucleo-cyphal/source/CyphalApp.cpp` — `ServiceDispatcherInit()` registers the GetInfo request+response ports; `OnReceiveUdp()` routes UDP datagrams to the dispatcher vs. the subject subscription based on `metadata->destination_address` (service multicast vs subject multicast); `ServiceResponseHandler()` / `GetInfoResponseHandler()`. -- `applications/nucleo-cyphal/include/GetInfoScanner.hpp` + `source/GetInfoScanner.cpp` — the host-testable scan window / transfer-ID tracker pattern to imitate for any client-side work. -- Tests: `applications/nucleo-cyphal/tests/` (`catch2-cyphal-getinfo-server.cpp`, `-client.cpp`, `-heartbeat.cpp`, `-record.cpp`), registered via `host_unit_test(...)` in `tests/CMakeLists.txt`. - -## Design findings - -### DSDL codegen already covers the new services - -`external/CMakeLists.txt` `generate_cyphal_dsdl()` runs nunavut over the whole -`third-party/public_regulated_data_types/uavcan/` tree, so the headers already -exist in the build dir (example: `build/native-llvm/generated/cyphal-dsdl/`): - -- `uavcan/node/ExecuteCommand_1_0.h` .. `_1_3.h` — **use 1.3** (the 1.0/1.1/1.2 - are `@deprecated`; 1.3 adds `COMMAND_IDENTIFY = 65529` and an `output` field in - the response). Request: `uint16 command` + `parameter` (max 112 bytes). Response: - `uint8 status` + `output`. -- `uavcan/_register/` — **note the namespace mangles to `_register`** because - `register` is a C++ keyword: - - `Access_1_0.h` → `uavcan_register_Access_Request_1_0` (name + value), - `uavcan_register_Access_Response_1_0` (timestamp, `_mutable`, `persistent`, value). - - `List_1_0.h` → request is just `uint16 index`; response is `uavcan_register_Name_1_0 name`. - - `Name_1_0.h` → `{uint8 elements[256]; size_t count;}`. - - `Value_1_0.h` → big C union (15 tags, max field 258 bytes), `_tag_` byte. -- Extents: Access req 515 B, Access resp 267 B, List resp 256 B, Value 259 B. - -Include them as `#include "uavcan/_register/Access_1_0.h"` etc. - -### RPC dispatcher handles multiple service ports - -`udpardRxRPCDispatcherReceive` (libudpard v1.x, `third-party/libudpard/libudpard/udpard.h:1056`) -returns `>0` on a completed transfer and the transfer carries `service_id` and -`is_request`. So we register several request ports (384, 385, 435) via -`udpardRxRPCDispatcherListen(&dispatcher, &port, id, /*is_request=*/true, extent)` -and dispatch on `transfer.service_id`. This is the extension point for the -current single-GetInfo `ServiceResponseHandler` (CyphalApp.cpp:588) — it must -become a switch over service-id. - -The service multicast group + `HyphaIpPrepareUdpReceive` are already set up in -`ServiceDispatcherInit()`; no new multicast memberships needed. - -### RegisterStore — new host-testable component - -Plan: a static-table register store following the `GetInfoScanner` precedent -(no target deps, host unit tests with Catch2): - -- Fixed array of entries: name buffer + `uavcan_register_Value_1_0` value + - `mutable`/`persistent` flags. Add `RegisterStore` in - `applications/nucleo-cyphal/include/RegisterStore.hpp` (+ source). -- API sketch: `Count()`, `NameAt(index, out_name)` (empty when OOB — terminates - List iteration), `Access(name, write_value, out_value, out_mutable, out_persistent)` - implementing the spec's write-then-read semantics: write skipped when value is - `empty` (tag 0) or register immutable; unknown name → empty value + cleared flags. -- Registers to expose (standard names from the 384.Access dsdl doc): - - `uavcan.node.id` → natural16[1] = {103} (matches `CyphalApp::NodeId`). - - `uavcan.node.description` → string, e.g. "nucleo-cyphal STM32H753ZI". - - `uavcan.pub.heartbeat.id` = 7509, `uavcan.srv.getinfo.id` = 430, - `uavcan.srv.executecommand.id` = 435, `uavcan.srv.register_access.id` = 384, - `uavcan.srv.register_list.id` = 385. -- Note: no NVS on the board yet → `persistent` flags will be claims only; either - keep all non-persistent or document the gap in PLAN.md. `COMMAND_STORE_PERSISTENT_STATES` - can respond SUCCESS as a no-op. - -### ExecuteCommand server — what to support - -- `COMMAND_RESTART` (65535): **no reset mechanism found yet** — grep for - `NVIC_SystemReset`/`__DSB`/`SystemReset` across `modules/` returned nothing. - Check `boards/nucleo_h753zi/include/BoardContext.hpp` (there is a `cortex` - module); if none, either add a `cortex::system::Reset()` or respond - `STATUS_BAD_STATE`/`STATUS_FAILURE` and log. **Investigate before wiring.** -- `COMMAND_STORE_PERSISTENT_STATES` (65530): SUCCESS no-op (no NVS). -- `COMMAND_EMERGENCY_STOP` (65531): could gate Heartbeat/Record publishing. -- `COMMAND_IDENTIFY` (65529): board has `status_pin_`/`error_pin_` GPIO pins in - `BoardContext` — could blink; optional. -- Unknown commands → `STATUS_BAD_COMMAND`. - -### Tests to write (host, Catch2) - -- `catch2-cyphal-executecommand.cpp` — request/response DSDL round-trip; command - enum serialization; response status values. -- `catch2-cyphal-register.cpp` — RegisterStore: Define/Count, NameAt bounds, - Access read vs write, immutable write rejected, empty-value write skipped, - unknown name → empty value. -- `catch2-cyphal-register-dsdl.cpp` (or fold in) — Access/List/Value round-trips, - string/union tag select/check helpers. -- Register the new tests in `applications/nucleo-cyphal/tests/CMakeLists.txt` with - `host_unit_test(NAME cyphal-... SOURCES ... LIBRARIES cyphal-dsdl CATCH2 - NO_CONFIGURATIONS NO_BOARDS)`. -- App-level service handlers behind hypha/udpard are not host-testable — add a - mock + test plan per AGENTS.md (same caveat as GetInfo). - -## Workflow reminders (from AGENTS.md) - -- Write a `PLAN.md`, review with human before committing. -- `gh issue` lookup/create and branch `issue-N` tracking `develop`; PR to `develop`. -- Build/tests: `cmake --workflow --preset on-host-native-llvm`, - `on-host-native-clang`; cross builds `on-target-cortex-m4-gcc-arm-none-eabi`, - `on-target-cortex-m7-gcc-arm-none-eabi`. Native-GCC preset does not work on - Darwin (see GOTCHAS.md). Run `./scripts/build-all-presets.sh` before PR. -- Update issue comments as progress is made. - -## Open questions for the human - -1. Reset mechanism for `COMMAND_RESTART` — is there a `cortex` reset API planned, - or should this first pass answer `STATUS_BAD_STATE` and log? -2. Which ExecuteCommand version to target — 1.3 (non-deprecated) is the strong - default unless yactui/tools expect 1.0. -3. Register set — confirm the port-ID registers list above is the right starting - set, and whether `uavcan.node.description` should be mutable in RAM only. diff --git a/PLAN.md b/PLAN.md deleted file mode 100644 index 0a0ed88..0000000 --- a/PLAN.md +++ /dev/null @@ -1,39 +0,0 @@ -# PLAN: Encapsulate libudpard in a `cyphal::Interface` implementation - -**Status:** In progress — steps 4–5 complete (socket abstraction + `CyphalUDPInterface` + GoogleTest suite); hypha adapter and app refactor remain. -**Issue:** https://github.com/emrainey/embedded-superloop/issues/61 -**Branch (when started):** `issue-61`, tracking `develop` - -## Summary - -Extract the libudpard plumbing inlined in `applications/nucleo-cyphal/source/CyphalApp.cpp` (~400 of 795 lines) into a reusable `jarnax::cyphal::CyphalUDPInterface` implementing `jarnax::cyphal::Interface`. Applications stop touching udpard directly. - -## Estimate - -4–7 days total including tests. - -## Steps - -1. Branch `issue-61` off `develop`. -2. [complete] Move the shared O1Heap pool into `jarnax-cyphal-udp`, implement `core::Allocator`, and provide libudpard memory-resource factories. -3. [complete] Add host-runnable GoogleTest coverage for the allocator and libudpard callbacks. -4. [complete] Socket abstraction: `source/include/jarnax/services/CyphalUDPSocket.hpp` (`udp::Endpoint`, `DatagramHandler`, `Socket` with Join/Leave/Send). Time comes from an injected `MicrosecondClock` (generic modules cannot include cortex headers). -5. [complete] `CyphalUDPInterface` in `source/services/CyphalUDPInterface.cpp` + internal header: all 6 `Interface` virtuals + `Loopable::Execute` TX drain + `DatagramHandler::OnDatagramReceived` RX dispatch; per-port transfer-ID counters; priority fixed Nominal; remembered request transfer-IDs for responses; fragment gather into scratch buffer; statistics. -6. Hypha adapter for the socket abstraction. -7. [complete] GoogleTest suite `tests/gtest-cyphal-udpinterface.cpp` with `tests/mocks/jarnax/services/MockUDPSocket.hpp`: Listen/Remove/IsListening lifecycle, Join-once semantics, Send publish/request/respond, TX error stats, RX subject round-trip, RPC request→response round-trip, unknown-group/bad-datagram handling. Valid RX datagrams are produced with a local `UdpardTx` producer (no hand-built wire formats). 62/62 pass on LLVM and AppleClang; both cross presets build. -8. Refactor `CyphalApp` onto `CyphalUDPInterface`; verify on hardware via pylink/RTT + yactui. -9. Run all presets and `./scripts/build-all-presets.sh`; PR against `develop`. - -## Key design decisions - -- Do **not** extend `Metadata` with priority/transfer-ID yet — priority is constant Nominal; per-port transfer-ID counters live inside the interface; response transfers echo the transfer-ID of the last remembered request from that client. -- Constructor takes `O1HeapPool&`, node-ID, `udp::Socket&`, and `MicrosecondClock&` — no singletons reached from inside; no cortex dependency so the generic module stays host-testable. -- Single redundant interface initially; `TransportStatistics` reports one interface entry. -- RX transfers are gathered into a contiguous scratch buffer bounded by `MaxExtent`; empty transfers are delivered as zero-length messages. -- Service ports share one RPC multicast group; Join is issued once for the first port and Leave when the last port is removed. - -## Acceptance criteria - -- No udpard types leak into application code. -- All host unit tests pass on LLVM and AppleClang; cross builds unbroken. -- nucleo-cyphal behaves identically on hardware. diff --git a/opencode.json b/opencode.json index 63571f3..8fdf1ee 100644 --- a/opencode.json +++ b/opencode.json @@ -19,6 +19,17 @@ } }, "mcp": { + "elf-mcp": { + "type": "local", + "command": [ + ".venv/bin/python", + "tools/elf-mcp/mcp_server.py" + ], + "cwd": ".", + "environment": { + "PYTHONPATH": "tools/elf-mcp" + } + }, "pylink-square-mcp": { "type": "local", "command": [ diff --git a/tools/elf-mcp/README.md b/tools/elf-mcp/README.md new file mode 100644 index 0000000..beb33a8 --- /dev/null +++ b/tools/elf-mcp/README.md @@ -0,0 +1,83 @@ +# elf-mcp + +An MCP server that turns the ARM GNU binutils and GNU ld link-maps into +structured-JSON tools for firmware flash/RAM analysis. It is the smallest +possible bridge between an AI agent and the on-target firmware artifacts: +given a firmware ELF it can report exact image sizes, linker-memory fill, +per-object bloat (from the `.map`), the largest symbols, whether the image +uses dynamic memory, and disassembly / DWARF address resolution. + +Sibling server [`../pylink-square-mcp`](../pylink-square-mcp) does live +J-Link flashing/debugging; this server is purely host-side (no target +connection needed). + +## Layout + +- `mcp_server.py` — JSON-RPC 2.0 over stdio entrypoint registered in + `opencode.json` (mirrors `pylink-square-mcp`'s protocol). +- `tools.py` — the nine tool handlers + the `TOOLS` registry. +- `parsers.py` — pure text parsers (no subprocesses): objdump/readelf/nm + output, linker-script `MEMORY{}`, `.map` memory config, per-object blame, + discarded input sections. +- `toolchain.py` — resolves `arm-none-eabi-{size,nm,readelf,objdump,addr2line}` + (`/Applications/ArmGNUToolchain/**`, `/opt/homebrew/bin`, `$PATH`). +- `tests/` — pytest suite; runs on any host with no ARM toolchain (tools are + exercised through a scripted fake runner). + +## Tools + +| Tool | What it answers | +| --- | --- | +| `list_elf_files` | Which firmware ELFs/link-maps exist under `build/`. | +| `binary_size` | Flash vs RAM totals with per-section classification. | +| `section_breakdown` | Verbose `readelf -S` listing (debug sections too). | +| `memory_map` | Fill percentage of each linker `MEMORY` region. | +| `largest_symbols` | Top-N symbols by size (`nm`), for bloat hunting. | +| `per_object_map` | Per-object flash/RAM contributions from the `.map`. | +| `verify_no_heap` | Scan for dynamic memory / C++ exception / static-init machinery. | +| `disassemble` | Disassemble a function or address range. | +| `addr2line` | Resolve address(es) -> function + source line (pairs with J-Link backtraces). | + +## Usage + +```jsonc +// opencode.json (already wired) +"elf-mcp": { + "type": "local", + "command": [".venv/bin/python", "tools/elf-mcp/mcp_server.py"], + "cwd": ".", + "environment": { "PYTHONPATH": "tools/elf-mcp" } +} +``` + +Restart opencode after editing `opencode.json` (config is not hot-reloaded). + +Each tool can also be driven from a shell: + +```sh +.venv/bin/python tools/elf-mcp/tools.py binary_size \ + --json '{"elf":"build/cortex-m7-gcc-arm-none-eabi/applications/nucleo-demo/firmware-nucleo-demo-basic-nucleo_h753zi.elf"}' +``` + +## Tests + +```sh +PYTHONPATH=tools/elf-mcp .venv/bin/python -m pytest tools/elf-mcp/tests -q +``` + +No cross-compiled targets or toolchain are required: parsers eat fixture text +and the tool handlers run against a scripted fake binutils runner. + +## Validation notes + +Against `firmware-nucleo-demo-basic-nucleo_h753zi.elf` (GCC 13.2, M7): + +- `binary_size` matches `arm-none-eabi-size` exactly: flash `0xe0f4`, + RAM `0x37b20`. +- `per_object_map` attributes input objects to flash within ~2% of the image + total. Its RAM figure excludes linker-reserved regions (`.process_stack`, + `.main_stack`, `.ethernet_dma_buffers`, `.dma_buffers`) and `.data` LMA + copies by design — use `binary_size` for exact totals. +- `.map` parsing tolerates both classic (`name 0xaddr 0xsize obj`) and + function-sections (bare header + raw leaf) layouts, and strips + `Discarded input sections` blocks wherever the linker printed them. \ No newline at end of file diff --git a/tools/elf-mcp/mcp_server.py b/tools/elf-mcp/mcp_server.py new file mode 100644 index 0000000..693d7ad --- /dev/null +++ b/tools/elf-mcp/mcp_server.py @@ -0,0 +1,130 @@ +"""elf-mcp: embeddded flash/RAM analysis MCP server (JSON-RPC 2.0 over stdio). + +Exposes arm binutils + linker-script/.map analysis as structured-JSON tools so +an AI agent can inspect firmware size, regions, per-object bloat, heap usage and +code addresses. Mirrors the stdio protocol of tools/pylink-square-mcp. +""" + +from __future__ import annotations + +import json +import os +import sys +import traceback + +try: + from tools import TOOLS +except ImportError: + sys.path.append(os.path.dirname(os.path.abspath(__file__))) + from tools import TOOLS + + +def log(msg: str) -> None: + sys.stderr.write(f"[elf-mcp] {msg}\n") + sys.stderr.flush() + + +def _reply(response: dict) -> None: + sys.stdout.write(json.dumps(response) + "\n") + sys.stdout.flush() + + +def main() -> None: + log("elf-mcp server starting...") + log(f"Exposing {len(TOOLS)} tools.") + + for line in sys.stdin: + if not line.strip(): + continue + try: + request = json.loads(line) + except Exception as exc: # noqa: BLE001 - keep the loop alive + log(f"Failed to parse JSON: {line.strip()}. Error: {exc}") + continue + + method = request.get("method") + req_id = request.get("id") + + if method == "initialize": + _reply({ + "jsonrpc": "2.0", + "id": req_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "elf-mcp", "version": "0.1.0"}, + }, + }) + log("Initialized protocol session.") + + elif method == "notifications/initialized": + log("Received notifications/initialized from client.") + + elif method == "tools/list": + tools_list = [ + {"name": t["name"], "description": t["description"], + "inputSchema": t["inputSchema"]} + for t in TOOLS + ] + _reply({ + "jsonrpc": "2.0", + "id": req_id, + "result": {"tools": tools_list}, + }) + log(f"Listed {len(tools_list)} tools.") + + elif method == "tools/call": + params = request.get("params", {}) + name = params.get("name") + arguments = params.get("arguments", {}) + log(f"Calling tool '{name}'...") + + tool = next((t for t in TOOLS if t["name"] == name), None) + if not tool: + _reply({ + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32601, "message": f"Method not found: {name}"}, + }) + else: + try: + exit_code, output = tool["handler"](arguments) + response = { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": output}], + "isError": exit_code != 0, + }, + } + except Exception as exc: # noqa: BLE001 + log(f"Error handling tool '{name}': {exc}\n{traceback.format_exc()}") + response = { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{ + "type": "text", + "text": f"Internal error executing tool '{name}': " + f"{exc}\n{traceback.format_exc()}", + }], + "isError": True, + }, + } + _reply(response) + log(f"Tool '{name}' finished execution.") + + else: + if req_id is not None: + _reply({ + "jsonrpc": "2.0", + "id": req_id, + "error": { + "code": -32601, + "message": f"Unsupported or unknown method: {method}", + }, + }) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/tools/elf-mcp/parsers.py b/tools/elf-mcp/parsers.py new file mode 100644 index 0000000..e544f94 --- /dev/null +++ b/tools/elf-mcp/parsers.py @@ -0,0 +1,656 @@ +"""Pure text parsers for ARM binutils / GNU ld output and linker scripts. + +These functions take *strings* (the stdout of a tool, or the text of a linker +script / map file) and return plain-Python data structures. They never spawn +a subprocess, so they are fully unit-testable and reusable by the MCP tools. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import Iterable, Optional + +# --------------------------------------------------------------------------- +# shared +# --------------------------------------------------------------------------- + + +def _parse_size(value: str) -> int: + """Parse a GNU size token like ``0x20000``, ``2M``, ``128K``, ``100000``. + + Also accepts a bare symbol name (resolved against ``symbols``) or a simple + ``A * N`` / ``A / N`` expression, matching how real linker scripts compute + region lengths (e.g. ``LENGTH = DTCM_SIZE / 2``). + """ + value = value.strip() + suffix = 1 + tail = value + for marker, mult in (("K", 1 << 10), ("M", 1 << 20), ("G", 1 << 30)): + if tail.upper().endswith(marker): + suffix = mult + tail = tail[:-1] + break + try: + return int(tail, 0) * suffix + except ValueError: + pass + # simple binary arithmetic: "A op B" where A/B are numbers or symbol refs + binary = re.match(r"^\s*(\S+)\s*([*/])\s*(\S+)\s*$", tail) + if binary: + try: + lhs = _parse_size(binary.group(1)) + rhs = _parse_size(binary.group(3)) + except ValueError: + raise + return (lhs * rhs) if binary.group(2) == "*" else (lhs // rhs) + raise ValueError(f"cannot parse numeric token {value!r}") + + +@dataclass(frozen=True) +class Section: + """One alloc section parsed from ``objdump -h``.""" + + name: str + size: int + vma: int + lma: int + flags: frozenset[str] = frozenset() + + @property + def is_alloc(self) -> bool: + return "ALLOC" in self.flags + + @property + def is_writable(self) -> bool: + return "WRITE" in self.flags # objdump shows READONLY; writable = not read-only + + @property + def is_stored(self) -> bool: + return "CONTENTS" in self.flags # has bytes in the image (vs NOBITS) + + @property + def is_readonly(self) -> bool: + return "READONLY" in self.flags + + +# --------------------------------------------------------------------------- +# objdump -h +# --------------------------------------------------------------------------- + +_OBJDUMP_SECTION_RE = re.compile( + r"^\s*(\d+)\s+(\S+)\s+([0-9a-fA-F]+)\s+([0-9a-fA-F]+)\s+([0-9a-fA-F]+)\s+\S+\s+\S+\s*$" +) + + +def parse_objdump_sections(text: str) -> list[Section]: + """Parse ``objdump -h`` output into a list of ``Section``. + + The flags for a section appear on the *following* line; we attach them to + the preceding header row. + """ + sections: list[Section] = [] + pending: Section | None = None + for raw in text.splitlines(): + line = raw.strip() + if not line: + pending = None + continue + match = _OBJDUMP_SECTION_RE.match(line) + if match: + pending = Section( + name=match.group(2), + size=int(match.group(3), 16), + vma=int(match.group(4), 16), + lma=int(match.group(5), 16), + flags=frozenset(), + ) + sections.append(pending) + continue + if pending is not None: + # Flag line, e.g. "CONTENTS, ALLOC, LOAD, READONLY, CODE" + new_flags = frozenset( + f.strip() for f in line.split(",") if f.strip() + ) | set(pending.flags) + pending = Section( + name=pending.name, + size=pending.size, + vma=pending.vma, + lma=pending.lma, + flags=new_flags, + ) + sections[-1] = pending + return [s for s in sections if s.is_alloc] + + +@dataclass(frozen=True) +class SizeBreakdown: + """Flash/RAM totals for an image.""" + + flash: int = 0 + ram: int = 0 + #: Name -> breakdown of the flash/RAM space that section occupies. + sections: tuple[tuple[str, int, int], ...] = () + + @property + def total(self) -> int: + return self.flash + self.ram + + def as_dict(self) -> dict: + return {"flash": self.flash, "ram": self.ram, "total": self.total} + + +def classify_section(section: Section) -> None | str: + """Classify an alloc section as ``"flash"``, ``"ram"`` or ``"other"``. + + - Read-only stored sections (``.text``, ``.rodata``, ``.tables`` ...) live + in flash. + - Stored writable sections (``.data``) have an LMA copy in flash **and** a + runtime VMA in RAM. + - Unstored writable sections (``.bss``, stacks, DMA buffers) live in RAM. + """ + if not section.is_alloc: + return "other" + if section.is_readonly: + return "flash" if section.size else "other" + if section.is_stored: + return "data" # both flash (LMA) and ram (VMA) + return "ram" + + +def compute_size_breakdown(sections: Iterable[Section]) -> SizeBreakdown: + """Compute flash/RAM totals for a list of alloc sections. + + ``.data``-like sections are charged to flash (LMA copy) and RAM (VMA). + """ + flash = 0 + ram = 0 + rows: list[tuple[str, int, int]] = [] + for s in sections: + kind = classify_section(s) + if kind == "flash": + flash += s.size + rows.append((s.name, s.size, 0)) + elif kind == "data": + flash += s.size + ram += s.size + rows.append((s.name, s.size, s.size)) + elif kind == "ram": + ram += s.size + rows.append((s.name, 0, s.size)) + rows.sort(key=lambda r: r[1] + r[2], reverse=True) + return SizeBreakdown(flash=flash, ram=ram, sections=tuple(rows)) + + +# --------------------------------------------------------------------------- +# nm -S --size-sort +# --------------------------------------------------------------------------- + +@dataclass(frozen=True) +class Symbol: + """One symbol from ``nm``.""" + + name: str + address: int | None + size: int | None + type: str + + @property + def kind(self) -> str: + if self.type in "Tt": + return "text" + if self.type in "Rr": + return "rodata" + if self.type in "Dd": + return "data" + if self.type in "Bb": + return "bss" + return "other" + + +_NM_LINE_RE = re.compile( + r"^\s*([0-9a-fA-F]+)\s+([0-9a-fA-F]+)\s+([A-Za-z])\s+(.+)$" +) +_NM_UNDEF_RE = re.compile(r"^\s*([A-Za-z])\s+(.+)$") + + +def parse_nm_symbols(text: str) -> list[Symbol]: + """Parse ``nm -S --size-sort`` output (debug/diagnostic lines tolerated).""" + symbols: list[Symbol] = [] + for raw in text.splitlines(): + line = raw.strip() + if not line: + continue + match = _NM_LINE_RE.match(line) + if match: + symbols.append( + Symbol( + name=match.group(4).strip(), + address=int(match.group(1), 16), + size=int(match.group(2), 16), + type=match.group(3), + ) + ) + continue + match = _NM_UNDEF_RE.match(line) + if match and len(line.split()) == 2: + symbols.append( + Symbol(name=match.group(2).strip(), address=None, size=None, + type=match.group(1)) + ) + return symbols + + +# --------------------------------------------------------------------------- +# readelf -S -W +# --------------------------------------------------------------------------- + +@dataclass(frozen=True) +class ReadelfSection: + number: int + name: str + type: str + addr: int + offset: int + size: int + flags: str + + +_READELF_SECTION_RE = re.compile( + r"^\s*\[\s*(\d+)\]\s+(\S*?)\s+(\S+)\s+([0-9a-fA-F]+)\s+([0-9a-fA-F]+)\s+" + r"([0-9a-fA-F]+)\s+([0-9a-fA-F]+)\s+(\S+)" +) + + +def parse_readelf_sections(text: str) -> list[ReadelfSection]: + """Parse ``readelf -S -W`` output (wide). + + Only row-shaped entries are kept; the leading NULL section (whose Name + column is empty) is skipped. + """ + out: list[ReadelfSection] = [] + for raw in text.splitlines(): + line = raw.strip() + if not line or not line.startswith("["): + continue + match = _READELF_SECTION_RE.match(line) + if not match: + continue + number = int(match.group(1)) + if number == 0: # readelf's resident NULL section + continue + name = match.group(2) + if not name: + continue + out.append( + ReadelfSection( + number=number, + name=name, + type=match.group(3), + addr=int(match.group(4), 16), + offset=int(match.group(5), 16), + size=int(match.group(6), 16), + flags=match.group(8), + ) + ) + return out + + +# --------------------------------------------------------------------------- +# linker script: MEMORY { } +# --------------------------------------------------------------------------- + +@dataclass(frozen=True) +class MemoryRegion: + name: str + attributes: str + origin: int + length: int + #: Where the region was defined (file:line), if known. + defines: str = "" + + @property + def is_writable(self) -> bool: + """A region is RAM-ish if its access attributes permit writes.""" + if self.attributes: + return "w" in self.attributes.lower() + return True # default attributes are rwx + + +_MEMORY_BLOCK_RE = re.compile(r"MEMORY\s*\{(.*?)\}", re.DOTALL) +_REGION_RE = re.compile( + r"^\s*(?P[A-Za-z_][A-Za-z0-9_]*)\s*(?:\((?P[^)]*)\))?\s*:" + r"\s*ORIGIN\s*=\s*(?P[^,\n]+)\s*,\s*LENGTH\s*=\s*(?P[^,\n]+)", + re.MULTILINE, +) +_ASSIGN_RE = re.compile(r"^\s*(\w+)\s*=\s*([^;]+);", re.MULTILINE) + + +def _parse_symbols(text: str) -> dict[str, str]: + """Collect top-level ``NAME = value;`` assignments from a linker script.""" + out: dict[str, str] = {} + for match in _ASSIGN_RE.finditer(text): + out[match.group(1)] = match.group(2).strip() + return out + + +def _resolve_token(token: str, symbols: dict[str, str]) -> int: + """Resolve a region ORIGIN/LENGTH token to an integer. + + Falls back through: numeric literal, top-level linker-script symbols + (expanded inside a simple ``A op B`` expression too), then a plain + ``A op B`` expression made of numeric literals. + """ + token = token.strip() + expanded = token + for name, value in symbols.items(): + expanded = re.sub(rf"\b{re.escape(name)}\b", value, expanded) + try: + return _parse_size(expanded) + except ValueError: + raise ValueError(f"cannot resolve {token!r} as an ORIGIN/LENGTH value") + + +def parse_linker_memory(text: str) -> list[MemoryRegion]: + """Parse the ``MEMORY { ... }`` block of a linker script. + + Handles ``LENGTH = 2M`` style literals as well as symbol references like + ``LENGTH = FLASH_SIZE`` (resolved from top-level assignments). + """ + symbols = _parse_symbols(text) + regions: list[MemoryRegion] = [] + match = _MEMORY_BLOCK_RE.search(text) + block = match.group(1) if match else "" + if not block: + return regions + for region_match in _REGION_RE.finditer(block): + attrs = region_match.group("attrs") or "" + try: + origin = _resolve_token(region_match.group("origin"), symbols) + length = _resolve_token(region_match.group("length"), symbols) + except ValueError: + continue + regions.append( + MemoryRegion( + name=region_match.group("name"), + attributes="".join(attrs.split()), + origin=origin, + length=length, + ) + ) + return regions + + +def region_for_address(regions: Iterable[MemoryRegion], addr: int) -> MemoryRegion | None: + """Return the first memory region containing ``addr``, or ``None``.""" + for region in regions: + if region.origin <= addr < region.origin + region.length: + return region + return None + + +# --------------------------------------------------------------------------- +# linker map file (+ per-object accounting) +# --------------------------------------------------------------------------- + +@dataclass +class MapSection: + """A top-level section seen in the map file.""" + + name: str + address: int + size: int + + +@dataclass +class ObjectBlame: + """Flash/RAM bytes attributable to one object file.""" + + object: str + flash: int = 0 + ram: int = 0 + sections: dict[str, int] = field(default_factory=dict) + + @property + def total(self) -> int: + return self.flash + self.ram + + +_MEMCONFIG_BLOCK_RE = re.compile( + r"Memory Configuration(.*?)\n\n", re.DOTALL +) +_STACK_RE = re.compile( + r"^\s*(?P[A-Za-z_][A-Za-z0-9_]*)\s+(?P0x[0-9a-fA-F]+)" + r"\s+(?P0x[0-9a-fA-F]+)\s+(?P\S*)" +) +#: Section-header row: either a bare input-section header (function-sections +#: style, e.g. " .text._ZN13GlobalContext10InitializeEv") or an output-section +#: total (".text 0x08000400 0xbc88"). Never has a trailing object path. +_MAP_HEADER_RE = re.compile( + r"^\s*(?P\.[A-Za-z0-9_.]+|[A-Za-z0-9_]+)" + r"(?:\s+0x(?P[0-9a-fA-F]{8,16})\s+0x(?P[0-9a-fA-F]+))?\s*$" +) +#: Classic leaf with a name, e.g. " .text 0x08000400 0x21c ". +_MAP_LEAF_RE = re.compile( + r"^\s*(?P\.?[A-Za-z_][A-Za-z0-9_.]*?)\s+0x(?P[0-9a-fA-F]{8,16})" + r"\s+0x(?P[0-9a-fA-F]+)\s+(?P\S.*)$" +) +#: Raw-leaf where the section header and its contribution are on separate +#: lines, e.g. " 0x08000650 0xe4 ". +_MAP_RAW_LEAF_RE = re.compile( + r"^\s*0x(?P[0-9a-fA-F]{8,16})\s+0x(?P[0-9a-fA-F]+)\s+(?P\S.*)$" +) +#: A row inside a ``Discarded input sections`` listing: either a named leaf +#: (" .text 0x00000000 0x0 "), a bare input-section header +#: (" .text._ZN13GlobalContext10InitializeEv") with a raw leaf below it, a +#: raw " 0x00000000 0x4 " line, or a plain value row (" 0x00000000 x"). +_DISCARD_ROW_RE = re.compile( + r"^\s*(?:" + r"\.?\w[\w.\-$]*$" # bare section header + r"|\.\S+\s+0x[0-9a-fA-F]+\s+0x[0-9a-fA-F]+\s+\S.*$" # named discard row + r"|0x[0-9a-fA-F]+\s+0x[0-9a-fA-F]+\s+\S.*$" # raw discard leaf + r"|0x[0-9a-fA-F]+\s+\S.*$" # symbol/value row + r")" +) + + +def _strip_discarded(text: str) -> str: + """Remove every ``Discarded input sections`` block from a map. + + The block may appear *anywhere* (some ld versions print it before the + memory configuration); it ends at the first line that is not a + name/addr/size/object discard row. + """ + out: list[str] = [] + discarding = False + for line in text.splitlines(): + if "Discarded input sections" in line: + discarding = True + continue + if not discarding: + out.append(line) + continue + if line.strip() and not _DISCARD_ROW_RE.match(line): + discarding = False + out.append(line) + return "\n".join(out) + + +def parse_map_memory_config(text: str) -> list[MemoryRegion]: + """Parse the ``Memory Configuration`` table printed in a GNU ld ``.map``. + + Returns an empty list if the table is absent (e.g. the map was made with a + plain ``-Map`` and no MEMORY command in the script); callers then fall back + to section-name heuristics. + """ + regions: list[MemoryRegion] = [] + lines = text.splitlines() + start = next( + (i for i, line in enumerate(lines) if "Memory Configuration" in line), None + ) + if start is None: + return regions + # The header row "Name Origin Length Attributes" is on its own line. + header = next( + (i for i in range(start + 1, len(lines)) if "Name" in lines[i]), None + ) + if header is None: + return regions + for line in lines[header + 1:]: + m = _STACK_RE.match(line) + if not m: + break # e.g. "*default*" or a blank line ends the table + try: + origin = int(m.group("origin"), 16) + length = int(m.group("length"), 16) + except ValueError: + continue + regions.append( + MemoryRegion( + name=m.group("name"), + attributes="".join(m.group("attrs").split()), + origin=origin, + length=length, + ) + ) + return regions + + +def _is_meta_section(name: str) -> bool: + """True for sections that never occupy device flash/RAM. + + Debug, symtab/strtab, group, vendor attribute and relocation sections are + link-time metadata (or section-group overviews) and must not be charged to + an object's flash/RAM totals. + """ + lowered = name.lower() + if lowered.startswith(".debug"): + return True + for prefix in (".arm.", ".group", ".symtab", ".strtab", ".dynsym", ".dynstr", + ".comment", ".note", ".rela", ".rel.", ".got", ".sdata"): + if lowered.startswith(prefix): + return True + return False + + +def parse_map_per_object(text: str, regions: Iterable[MemoryRegion] = ()) -> list[ObjectBlame]: + """Aggregate the size of each input object from a GNU ld ``.map`` file. + + ``regions`` (from :func:`parse_map_memory_config` or the linker script) are + used to classify each leaf's address as flash vs ram; without them the + section name header is used (`text/d`/`bss` heuristics). + + Supports both the classic ``name 0xaddr 0xsize `` format and the + function-sections format where a bare ``name`` header line is followed by a + raw ``0xaddr 0xsize `` contribution line. ``Discarded input + sections`` blocks are removed wherever they appear so GC'd input does not + pollute totals. + """ + region_list = list(regions) + text = _strip_discarded(text) + blames: dict[str, ObjectBlame] = {} + current_section: MapSection | None = None + in_memconfig = False + + def _classify(addr: int, section_name: str) -> str: + if region_list: + region = region_for_address(region_list, addr) + if region is not None: + return "ram" if region.is_writable else "flash" + head = section_name + if head.startswith(".text") or head.startswith(".rodata"): + return "flash" + return "ram" + + for raw in text.splitlines(): + line = raw.strip() + if not line: + continue + if "Memory Configuration" in line: + in_memconfig = True + continue + if "Linker script and memory map" in line: + in_memconfig = False + continue + if in_memconfig: + # Region table rows ("ITCM ... xrw") must not count as object input. + continue + # Raw-leaf: " 0x08000650 0xe4 " (function-sections format). + raw_match = _MAP_RAW_LEAF_RE.match(raw) + if raw_match and current_section is not None: + if _is_meta_section(current_section.name): + continue + try: + addr = int(raw_match.group("addr"), 16) + size = int(raw_match.group("size"), 16) + except ValueError: + continue + obj = raw_match.group("obj").strip() + if obj and not obj.startswith("load address"): + blame = blames.setdefault(obj, ObjectBlame(object=obj)) + kind = _classify(addr, current_section.name) + if kind == "flash": + blame.flash += size + else: + blame.ram += size + blame.sections[current_section.name] = ( + blame.sections.get(current_section.name, 0) + size + ) + continue + # Symbol/value rows like "0x08000400 __aeabi_frsub" or + # "0x00000001 ASSERT(...)" — single-hex rows, never section contributions. + if line.startswith("0x"): + continue + bare_match = _MAP_HEADER_RE.match(raw) + leaf_match = _MAP_LEAF_RE.match(raw) + if bare_match and not leaf_match: + # Output-section header (".name 0xaddr 0xsize") or bare input-section + # header (".name") — either way it starts a new section context. + try: + current_section = MapSection( + name=bare_match.group("name"), + address=0, + size=0, + ) + except ValueError: + current_section = None + continue + if leaf_match and current_section is not None: + if _is_meta_section(leaf_match.group("name")): + continue + try: + addr = int(leaf_match.group("addr"), 16) + size = int(leaf_match.group("size"), 16) + except ValueError: + continue + leaf_name = leaf_match.group("name") + obj = leaf_match.group("obj").strip() + if obj and not obj.startswith("load address"): + blame = blames.setdefault(obj, ObjectBlame(object=obj)) + kind = _classify(addr, leaf_name) + if kind == "flash": + blame.flash += size + else: + blame.ram += size + blame.sections[leaf_name] = ( + blame.sections.get(leaf_name, 0) + size + ) + return sorted(blames.values(), key=lambda b: b.total, reverse=True) + + +def parse_map_discarded(text: str) -> list[str]: + """Extract the ``Discarded input sections`` listing (for dead-code eyeballing).""" + marker = "Discarded input sections" + if marker not in text: + return [] + tail = text.split(marker, 1)[1] + out: list[str] = [] + for raw in tail.splitlines(): + line = raw.strip() + if not line: + continue + if re.match(r"^\s*\.\.\.\s*$", line): + break + if re.match(r"^[\w./].*\s+0x[0-9a-fA-F]+", line): + out.append(line) + return out \ No newline at end of file diff --git a/tools/elf-mcp/tests/__init__.py b/tools/elf-mcp/tests/__init__.py new file mode 100644 index 0000000..1fcc2bf --- /dev/null +++ b/tools/elf-mcp/tests/__init__.py @@ -0,0 +1,5 @@ +"""Unit tests for tools/elf-mcp/parsers.py. + +Fixtures reproduce the *text* produced by GNU binutils and ld on the host; +no ARM toolchain or cross-compiled artifact is required to run these tests. +""" \ No newline at end of file diff --git a/tools/elf-mcp/tests/test_parsers.py b/tools/elf-mcp/tests/test_parsers.py new file mode 100644 index 0000000..cfdfcb0 --- /dev/null +++ b/tools/elf-mcp/tests/test_parsers.py @@ -0,0 +1,334 @@ +"""Unit tests for tools/elf-mcp/parsers.py.""" + +from parsers import ( + MemoryRegion, + _is_meta_section, + _parse_size, + _strip_discarded, + classify_section, + compute_size_breakdown, + parse_linker_memory, + parse_map_discarded, + parse_map_memory_config, + parse_map_per_object, + parse_nm_symbols, + parse_objdump_sections, + parse_readelf_sections, + region_for_address, + Section, +) + + +# --------------------------------------------------------------------------- +# shared / size tokens +# --------------------------------------------------------------------------- + + +def test_parse_size_literals(): + assert _parse_size("0x20000") == 0x20000 + assert _parse_size("2M") == 2 * 1024 * 1024 + assert _parse_size("128K") == 128 * 1024 + assert _parse_size("128k") == 128 * 1024 + assert _parse_size("1G") == 1 << 30 + assert _parse_size("100000") == 100000 + + +def test_parse_size_expression(): + assert _parse_size("2 * 1024") == 2048 + assert _parse_size("262144 / 2") == 131072 + assert _parse_size("64K / 4") == 16384 + + +# --------------------------------------------------------------------------- +# objdump -h +# --------------------------------------------------------------------------- + +OBJDUMP_H = """\ +firmware.elf: file format elf32-littlearm + +Sections: +Idx Name Size VMA LMA File off Algn + 0 .text 0000bc88 08000400 08000400 00010300 2**2 + CONTENTS, ALLOC, LOAD, READONLY, CODE + 1 .data 00000108 24000000 0800dd54 00020300 2**3 + CONTENTS, ALLOC, LOAD, DATA + 2 .bss 0000eca4 24000800 24000800 00030000 2**3 + ALLOC + 3 .debug_info 00000000 00000000 00000000 000c0000 2**0 + DEBUG +""" + + +def test_parse_objdump_sections_attaches_flags_and_filters(): + sections = parse_objdump_sections(OBJDUMP_H) + by_name = {s.name: s for s in sections} + assert set(by_name) == {".text", ".data", ".bss"} # .debug_info dropped + assert by_name[".text"].is_readonly + assert by_name[".text"].is_stored + assert by_name[".data"].is_stored and not by_name[".data"].is_readonly + assert not by_name[".bss"].is_stored + assert by_name[".text"].vma == 0x08000400 + assert by_name[".text"].lma == 0x08000400 + assert by_name[".data"].lma == 0x0800dd54 + + +def test_classify_section(): + readonly = Section(name=".text", size=4, vma=0, lma=0, + flags=frozenset(["CONTENTS", "ALLOC", "LOAD", "READONLY"])) + stored_wr = Section(name=".data", size=4, vma=0, lma=0, + flags=frozenset(["CONTENTS", "ALLOC", "LOAD"])) + bss = Section(name=".bss", size=4, vma=0, lma=0, + flags=frozenset(["ALLOC"])) + debug = Section(name=".debug_info", size=4, vma=0, lma=0, + flags=frozenset(["DEBUG"])) + assert classify_section(readonly) == "flash" + assert classify_section(stored_wr) == "data" + assert classify_section(bss) == "ram" + assert classify_section(debug) == "other" + + +def test_compute_size_breakdown(): + sections = parse_objdump_sections(OBJDUMP_H) + breakdown = compute_size_breakdown(sections) + assert breakdown.flash == 0xbc88 + 0x108 # text + data LMA copy + assert breakdown.ram == 0x108 + 0xeca4 # data VMA + bss + rows = dict((name, (f, r)) for name, f, r in breakdown.sections) + assert rows[".data"] == (0x108, 0x108) + + +# --------------------------------------------------------------------------- +# nm -S --size-sort --radix=x +# --------------------------------------------------------------------------- + +NM_TEXT = """\ +08000400 000000fc T _ZN4Foo3RunEv +08010000 00000020 R foo_format +24000000 00000400 D bar_counter +24001000 0000ffff B bar_buf + U malloc + U memset +nm: warning: some junk input +""" + + +def test_parse_nm_symbols(): + symbols = parse_nm_symbols(NM_TEXT) + by_name = {s.name: s for s in symbols} + assert by_name["_ZN4Foo3RunEv"].kind == "text" + assert by_name["foo_format"].kind == "rodata" + assert by_name["bar_counter"].kind == "data" + assert by_name["bar_buf"].size == 0xffff + assert by_name["malloc"].address is None + assert by_name["malloc"].size is None + assert by_name["memset"].type == "U" + + +# --------------------------------------------------------------------------- +# readelf -S -W +# --------------------------------------------------------------------------- + +READELF_S = """\ +Section Headers: + [Nr] Name Type Address Off Size ES Flg Lk Inf Al + [ 0] NULL 0000000000000000 000000 000000 00 0 0 0 + [ 1] .text PROGBITS 08000400 010300 00bc88 00 AX 0 0 4 + [ 2] .rodata PROGBITS 0800c0bc 01bfbc 001bc8 00 AX 0 0 4 + [ 3] .bss NOBITS 24000800 020000 000eca4 00 WA 0 0 8 + [ 4] .debug PROGBITS 00000000 030000 0089453 00 0 0 1 +""" + + +def test_parse_readelf_sections(): + sections = parse_readelf_sections(READELF_S) + assert len(sections) == 4 # NULL row (empty name) skipped + text = sections[0] + assert text.name == ".text" + assert text.type == "PROGBITS" + assert text.addr == 0x08000400 + assert text.offset == 0x010300 + assert text.size == 0xbc88 + assert text.flags == "AX" + assert sections[2].size == 0xeca4 + assert sections[3].name == ".debug" + + +# --------------------------------------------------------------------------- +# linker script MEMORY {} +# --------------------------------------------------------------------------- + +LD_MEMORY = """\ +FLASH_SIZE = 2M; +DTCM_SIZE = 128K; +SRAM1_SIZE = 512K; + +MEMORY +{ + ITCM (xrw) : ORIGIN = 0x00000000, LENGTH = 128K + FLASH (rx) : ORIGIN = 0x08000000, LENGTH = FLASH_SIZE + DTCM (xrw) : ORIGIN = 0x20000000, LENGTH = DTCM_SIZE + SRAM1 (xrw) : ORIGIN = 0x24000000, LENGTH = SRAM1_SIZE / 2 +} +""" + + +def test_parse_linker_memory(): + regions = parse_linker_memory(LD_MEMORY) + by_name = {r.name: r for r in regions} + assert set(by_name) == {"ITCM", "FLASH", "DTCM", "SRAM1"} + assert by_name["FLASH"].origin == 0x08000000 + assert by_name["FLASH"].length == 2 * 1024 * 1024 + assert not by_name["FLASH"].is_writable + assert by_name["DTCM"].length == 128 * 1024 + assert by_name["DTCM"].is_writable + assert by_name["SRAM1"].length == 256 * 1024 # expression resolved + assert by_name["ITCM"].attributes == "xrw" + + +def test_region_for_address(): + regions = parse_linker_memory(LD_MEMORY) + flash = region_for_address(regions, 0x08000400) + sram1 = region_for_address(regions, 0x24010000) + itcm = region_for_address(regions, 0x00001000) + assert flash is not None and flash.name == "FLASH" + assert sram1 is not None and sram1.name == "SRAM1" + assert itcm is not None and itcm.name == "ITCM" + assert region_for_address(regions, 0x24080000) is None # past SRAM1/2 + + +# --------------------------------------------------------------------------- +# map file: memory configuration +# --------------------------------------------------------------------------- + +MAP_MEMORY_CONFIG = """\ +Cross Reference Table + +Memory Configuration + +Name Origin Length Attributes +ITCM 0x0000000000000000 0x0000000000020000 xrw +FLASH 0x0000000008000000 0x0000000000200000 rx +DTCM 0x0000000002000000 0x0000000000020000 xrw +*default* 0x0000000000000000 0xffffffffffffffff + +Linker script and memory map +""" + + +def test_parse_map_memory_config(): + regions = parse_map_memory_config(MAP_MEMORY_CONFIG) + by_name = {r.name: r for r in regions} + assert set(by_name) == {"ITCM", "FLASH", "DTCM"} + assert by_name["FLASH"].origin == 0x08000000 + assert by_name["FLASH"].length == 0x00200000 + assert not by_name["FLASH"].is_writable + assert by_name["DTCM"].is_writable + + +# --------------------------------------------------------------------------- +# map file: per-object accounting +# --------------------------------------------------------------------------- + +MAP_FULL = """\ + 0x08000000 _start + 0x08000400 Reset_Handler + +Discarded input sections + + .text.startup._GLOBAL__sub_I_main + 0x00000000 0x4 dead_boot.o + .bss + 0x00000000 0x10 dead_boot.o + +Memory Configuration + +Name Origin Length Attributes +ITCM 0x0000000000000000 0x0000000000020000 xrw +FLASH 0x0000000008000000 0x0000000000200000 rx +DTCM 0x0000000002000000 0x0000000000020000 xrw +*default* 0x0000000000000000 0xffffffffffffffff + +Linker script and memory map + +.text 0x08000400 0xbc88 + .text 0x08000400 0x21c source/GlobalContext.cpp.obj + 0x0800061c _ZN13GlobalContext10InitializeEv + .text 0x08000650 0x4 source/utility.cpp.obj + .text._ZN4core10StateChartI9DemoStateE7RunOnceEv + 0x08000654 0xfc core/statechart.cpp.obj + .rodata 0x0800c0bc 0x1bc8 + .rodata 0x0800c0bc 0x50 source/utility.cpp.obj +.bss 0x24000800 0xeca4 load address 0x0800e25c + .bss 0x24000800 0x14 source/app.cpp.obj + .bss 0x24000814 0xffff core/shared.cpp.obj +.debug_str 0x24001000 0x89453 source/GlobalContext.cpp.obj +""" + + +def test_parse_map_per_object_classic_and_function_sections(): + regions = parse_map_memory_config(MAP_FULL) + blames = parse_map_per_object(MAP_FULL, regions) + by_obj = {b.object: b for b in blames} + + # dead_boot.o lives only in the Discarded block -> must be absent. + assert "dead_boot.o" not in by_obj + + gc = by_obj["source/GlobalContext.cpp.obj"] + assert gc.flash == 0x21c + assert gc.ram == 0 # .debug_str meta row excluded + + util = by_obj["source/utility.cpp.obj"] + assert util.flash == 0x4 + 0x50 # .text + .rodata contributions + assert util.ram == 0 + + # function-sections bare header + raw leaf + sc = by_obj["core/statechart.cpp.obj"] + assert sc.flash == 0xfc + + # bss leaf in a RAM region (DTCM here) -> ram; ".bss ... load address" row skipped + app = by_obj["source/app.cpp.obj"] + assert app.flash == 0 + assert app.ram == 0x14 + + shared = by_obj["core/shared.cpp.obj"] + assert shared.ram == 0xffff + + # sorted by total descending + totals = [b.total for b in blames] + assert totals == sorted(totals, reverse=True) + + +def test_parse_map_per_object_without_regions_uses_names(): + blames = parse_map_per_object(MAP_FULL) + util = next(b for b in blames if b.object == "source/utility.cpp.obj") + assert util.flash == 0x54 # .text/.rodata heuristic + app = next(b for b in blames if b.object == "source/app.cpp.obj") + assert app.ram == 0x14 + + +def test_parse_map_discarded(): + discarded = parse_map_discarded(MAP_FULL) + assert any("dead_boot.o" in row for row in discarded) + + +def test_strip_discarded_removes_block(): + stripped = _strip_discarded(MAP_FULL) + assert "dead_boot.o" not in stripped + assert "Discarded input sections" not in stripped + assert "Memory Configuration" in stripped + + +# --------------------------------------------------------------------------- +# meta-section filter +# --------------------------------------------------------------------------- + + +def test_is_meta_section(): + assert _is_meta_section(".debug_info") + assert _is_meta_section(".debug_str") + assert _is_meta_section(".ARM.attributes") + assert _is_meta_section(".group") + assert _is_meta_section(".symtab") + assert _is_meta_section(".rela.dyn") + assert not _is_meta_section(".text") + assert not _is_meta_section(".bss._Z4main") + assert not _is_meta_section(".rodata._ZTV3Foo") \ No newline at end of file diff --git a/tools/elf-mcp/tests/test_tools.py b/tools/elf-mcp/tests/test_tools.py new file mode 100644 index 0000000..4e41aeb --- /dev/null +++ b/tools/elf-mcp/tests/test_tools.py @@ -0,0 +1,380 @@ +"""Unit tests for tools/elf-mcp/tools.py. + +Uses a scripted FakeRunner instead of real binutils so the suite runs on any +host (no ARM toolchain needed) and every expected subprocess call is asserted. +""" + +from pathlib import Path + +import pytest + +from toolchain import CommandResult +from tools import PROJECT_ROOT, TOOLS + + +def _handler(name): + tool = next(t for t in TOOLS if t["name"] == name) + return tool["handler"] + + +class FakeRunner: + """Records calls, fails on any unscripted invocation.""" + + def __init__(self): + self.calls = [] + self._script = {} + + def add(self, tool, args, *, returncode=0, stdout="", stderr=""): + self._script[(tool, tuple(args))] = CommandResult(returncode, stdout, stderr) + return self + + def expect(self, tool, args): + return (tool, list(args)) in [(t, a) for t, a in self.calls] + + def __call__(self, tool, args, *, cwd=None): + key = (tool, tuple(args)) + self.calls.append((tool, list(args))) + if key not in self._script: + raise AssertionError(f"unscripted call: {tool} {args}") + return self._script[key] + + +@pytest.fixture +def elf_file(tmp_path): + path = tmp_path / "firmware.elf" + path.write_bytes(b"\x7fELF") + return str(path) + + +# --------------------------------------------------------------------------- +# binary_size / section_breakdown / memory_map +# --------------------------------------------------------------------------- + +OBJDUMP_H = """\ +firmware.elf: file format elf32-littlearm + +Sections: +Idx Name Size VMA LMA File off Algn + 0 .text 0000bc88 08000400 08000400 00010300 2**2 + CONTENTS, ALLOC, LOAD, READONLY, CODE + 1 .data 00000108 24000000 0800dd54 00020300 2**3 + CONTENTS, ALLOC, LOAD, DATA + 2 .bss 0000eca4 20000000 20000000 00030000 2**3 + ALLOC + 3 .zero_table 00000038 0800dd54 0800dd54 00020300 2**3 + CONTENTS, ALLOC, LOAD, READONLY, DATA +""" + + +def test_binary_size(elf_file, tmp_path): + runner = FakeRunner().add("objdump", ["-h", elf_file], stdout=OBJDUMP_H) + exit_code, output = _handler("binary_size")( + {"elf": elf_file}, runner=runner, root=str(tmp_path) + ) + assert exit_code == 0 + data = __import__("json").loads(output) + assert data["flash"] == 0xbc88 + 0x108 + 0x38 + assert data["ram"] == 0x108 + 0xeca4 + assert runner.expect("objdump", ["-h", elf_file]) + + +def test_section_breakdown(elf_file, tmp_path): + readelf = """\ +Section Headers: + [Nr] Name Type Address Off Size ES Flg Lk Inf Al + [ 1] .text PROGBITS 08000400 010300 00bc88 00 AX 0 0 4 + [ 3] .bss NOBITS 24000800 020000 000eca4 00 WA 0 0 8 +""" + runner = FakeRunner().add("readelf", ["-SW", elf_file], stdout=readelf) + exit_code, output = _handler("section_breakdown")( + {"elf": elf_file}, runner=runner, root=str(tmp_path) + ) + assert exit_code == 0 + data = __import__("json").loads(output) + assert data["count"] == 2 + assert data["sections"][1]["size"] == 0xeca4 + + +LD_MEMORY = """\ +MEMORY +{ + FLASH (rx) : ORIGIN = 0x08000000, LENGTH = 2M + DTCM (xrw) : ORIGIN = 0x20000000, LENGTH = 128K +} +""" + + +def test_memory_map_with_explicit_ld(elf_file, tmp_path): + ld_path = tmp_path / "app.ld" + ld_path.write_text(LD_MEMORY) + runner = FakeRunner().add("objdump", ["-h", elf_file], stdout=OBJDUMP_H) + exit_code, output = _handler("memory_map")( + {"elf": elf_file, "ld": str(ld_path)}, runner=runner, root=str(tmp_path) + ) + assert exit_code == 0 + data = __import__("json").loads(output) + regions = {r["region"]: r for r in data["regions"]} + # .text 0xbc88 + .data 0x108 + .zero_table 0x38 inside FLASH(0x08000000..+2M) + assert regions["FLASH"]["used"] == 0xbc88 + 0x108 + 0x38 + assert regions["DTCM"]["used"] == 0xeca4 + assert regions["FLASH"]["fill_percent"] == pytest.approx( + 100.0 * (0xbc88 + 0x108 + 0x38) / (2 * 1024 * 1024), abs=0.01 + ) + + +def test_memory_map_missing_regions_is_error(elf_file, tmp_path): + # No ld and no map sibling -> no regions -> error result, exit 1. + runner = FakeRunner().add("objdump", ["-h", elf_file], stdout=OBJDUMP_H) + exit_code, output = _handler("memory_map")( + {"elf": elf_file}, runner=runner, root=str(tmp_path) + ) + assert exit_code == 1 + data = __import__("json").loads(output) + assert "MEMORY" in data["error"] + + +# --------------------------------------------------------------------------- +# largest_symbols +# --------------------------------------------------------------------------- + +NM_SORTED = """\ +24001000 0000ffff B bar_buf +08000400 000000fc T _ZN4Foo3RunEv +08010000 00000020 R foo_format +""" + + +def test_largest_symbols(elf_file, tmp_path): + runner = FakeRunner().add( + "nm", ["-S", "--size-sort", "--radix=x", elf_file], stdout=NM_SORTED + ) + exit_code, output = _handler("largest_symbols")( + {"elf": elf_file, "count": 2}, runner=runner, root=str(tmp_path) + ) + assert exit_code == 0 + data = __import__("json").loads(output) + assert [s["name"] for s in data["symbols"]] == ["bar_buf", "_ZN4Foo3RunEv"] + assert data["total_text_bytes"] == 0xfc + + +# --------------------------------------------------------------------------- +# per_object_map +# --------------------------------------------------------------------------- + +MAP_TEXT = """\ +Memory Configuration + +Name Origin Length Attributes +ITCM 0x0000000000000000 0x0000000000020000 xrw +FLASH 0x0000000008000000 0x0000000000200000 rx +*default* 0x0000000000000000 0xffffffffffffffff + +Linker script and memory map + +.text 0x08000400 0xbc88 + .text 0x08000400 0x21c source/GlobalContext.cpp.obj + .text._ZN4core10StateChartI9DemoStateE7RunOnceEv + 0x08000654 0xfc core/statechart.cpp.obj +.bss 0x24000800 0xeca4 load address 0x0800e25c + .bss 0x24000800 0x14 source/app.cpp.obj +""" + + +def test_per_object_map(tmp_path): + map_path = tmp_path / "firmware.map" + map_path.write_text(MAP_TEXT) + exit_code, output = _handler("per_object_map")( + {"map": str(map_path), "top": 10}, runner=FakeRunner(), root=str(tmp_path) + ) + assert exit_code == 0 + data = __import__("json").loads(output) + by_obj = {o["object"]: o for o in data["objects"]} + assert by_obj["source/GlobalContext.cpp.obj"]["flash"] == 0x21c + assert by_obj["core/statechart.cpp.obj"]["flash"] == 0xfc + assert by_obj["source/app.cpp.obj"]["ram"] == 0x14 + # the "load address" pseudo-row contributes nothing + assert data["covered_flash"] == 0x21c + 0xfc + assert data["covered_ram"] == 0x14 + + +def test_per_object_map_requires_argument(): + exit_code, output = _handler("per_object_map")({}, runner=FakeRunner()) + assert exit_code == 1 + data = __import__("json").loads(output) + assert "required" in data["error"] + + +# --------------------------------------------------------------------------- +# verify_no_heap +# --------------------------------------------------------------------------- + +NM_CLEAN = """\ +08000400 000000fc T _ZN4Foo3RunEv +""" +NM_DIRTY = NM_CLEAN + """\ + U malloc +""" +DISASM_CLEAN = """\ +08000400 <_ZN4Foo3RunEv>: + 8000400: 4770 bx lr +""" + + +def test_verify_no_heap_clean(elf_file, tmp_path): + runner = FakeRunner() + runner.add("nm", ["--defined-only", elf_file], stdout=NM_CLEAN) + runner.add("objdump", ["-d", elf_file], stdout=DISASM_CLEAN) + exit_code, output = _handler("verify_no_heap")( + {"elf": elf_file}, runner=runner, root=str(tmp_path) + ) + assert exit_code == 0 + data = __import__("json").loads(output) + assert data["pass"] is True + assert data["violations"] == [] + + +def test_verify_no_heap_flags_malloc_symbol(elf_file, tmp_path): + runner = FakeRunner() + runner.add("nm", ["--defined-only", elf_file], stdout=NM_DIRTY) + runner.add("objdump", ["-d", elf_file], stdout=DISASM_CLEAN) + exit_code, output = _handler("verify_no_heap")( + {"elf": elf_file}, runner=runner, root=str(tmp_path) + ) + data = __import__("json").loads(output) + assert data["pass"] is False + assert any(v["symbol"] == "malloc" for v in data["violations"]) + + +# --------------------------------------------------------------------------- +# disassemble / addr2line +# --------------------------------------------------------------------------- + +NM_WITH_SIZE = """\ +08000400 000000fc T _ZN4Foo3RunEv +""" + + +def test_disassemble_by_symbol(elf_file, tmp_path): + runner = FakeRunner() + runner.add("nm", ["-S", elf_file], stdout=NM_WITH_SIZE) + disasm = "08000400 <_ZN4Foo3RunEv>:\n".replace("\n", "") + "__leading\n" + runner.add( + "objdump", + ["-d", "--no-show-raw-insn", "--start-address=0x8000400", + "--stop-address=0x80004fc", elf_file], + stdout=disasm, + ) + exit_code, output = _handler("disassemble")( + {"elf": elf_file, "symbol": "_ZN4Foo3RunEv"}, runner=runner, root=str(tmp_path) + ) + assert exit_code == 0 + data = __import__("json").loads(output) + assert data["start"] == 0x08000400 + assert data["size"] == 0xfc + assert "__leading" in data["disassembly"] + + +def test_disassemble_by_address(elf_file, tmp_path): + runner = FakeRunner() + runner.add( + "objdump", + ["-d", "--no-show-raw-insn", "--start-address=0x8000400", + "--stop-address=0x8000480", elf_file], + stdout="before\n08000400: 4770 bx lr\nafter\n", + ) + exit_code, output = _handler("disassemble")( + {"elf": elf_file, "address": "0x08000400", "span": 0x80}, + runner=runner, root=str(tmp_path), + ) + assert exit_code == 0 + data = __import__("json").loads(output) + assert data["address"] == 0x08000400 + assert "bx lr" in data["disassembly"] + + +def test_disassemble_unknown_symbol_is_error(elf_file, tmp_path): + runner = FakeRunner().add("nm", ["-S", elf_file], stdout=NM_WITH_SIZE) + exit_code, output = _handler("disassemble")( + {"elf": elf_file, "symbol": "NoSuch"}, runner=runner, root=str(tmp_path) + ) + assert exit_code == 1 + assert "not found" in __import__("json").loads(output)["error"] + + +def test_addr2line(elf_file, tmp_path, monkeypatch): + runner = FakeRunner() + runner.add("addr2line", ["-e", elf_file, "-f", "-i", "0x08000400"], + stdout="_ZN4Foo3RunEv\nsource/foo.cpp:41") + runner.add("addr2line", ["-e", elf_file, "-f", "-i", "0x08000800"], + stdout="_ZN4Bar4StopEv\nsource/bar.cpp:7") + exit_code, output = _handler("addr2line")( + {"elf": elf_file, "address": ["0x08000400", "0x08000800"]}, + runner=runner, root=str(tmp_path), + ) + assert exit_code == 0 + data = __import__("json").loads(output) + assert data["resolved"][0] == { + "address": "0x08000400", + "function": "_ZN4Foo3RunEv", + "source": "source/foo.cpp:41", + } + assert data["resolved"][1]["source"] == "source/bar.cpp:7" + + +# --------------------------------------------------------------------------- +# list_elf_files +# --------------------------------------------------------------------------- + + +def _scaffold_build(tmp_path): + root = tmp_path / "proj" + m7 = (root / "build" / "cortex-m7-gcc-arm-none-eabi" / "applications" / + "nucleo-demo") + m7.mkdir(parents=True) + (m7 / "firmware-nucleo-demo-basic-nucleo_h753zi.elf").write_bytes(b"\x7fELF") + (m7 / "firmware-nucleo-demo-basic-nucleo_h753zi.map").write_text("x") + m4 = (root / "build" / "cortex-m4-gcc-arm-none-eabi" / "applications" / + "emb-demo") + m4.mkdir(parents=True) + (m4 / "firmware-emb-demo-basic-stm32_f4ve_v2.elf").write_bytes(b"\x7fELF") + # a non-elf file and a .dot. copy should be ignored + (m4 / "notes.txt").write_text("noise") + (m4 / "firmware-emb-demo.min.dot").write_bytes(b"digraph{}") + return root + + +def test_list_elf_files(tmp_path): + root = _scaffold_build(tmp_path) + exit_code, output = _handler("list_elf_files")( + {"build_root": "build"}, runner=FakeRunner(), root=str(root) + ) + assert exit_code == 0 + data = __import__("json").loads(output) + assert data["count"] == 2 + elves = [a["relative"] for a in data["artifacts"]] + assert any("nucleo-demo" in e and e.endswith(".elf") for e in elves) + m7 = next(a for a in data["artifacts"] if "nucleo-demo" in a["relative"]) + assert m7["map"] is not None + m4 = next(a for a in data["artifacts"] if "emb-demo" in a["relative"]) + assert m4["map"] is None + + +def test_list_elf_files_preset_filter(tmp_path): + root = _scaffold_build(tmp_path) + exit_code, output = _handler("list_elf_files")( + {"build_root": "build", + "preset": "on-target-cortex-m7-gcc-arm-none-eabi"}, + runner=FakeRunner(), root=str(root), + ) + data = __import__("json").loads(output) + assert data["count"] == 1 + assert "nucleo-demo" in data["artifacts"][0]["relative"] + + +def test_list_elf_files_missing_root(): + exit_code, output = _handler("list_elf_files")( + {"build_root": "does-not-exist"}, + runner=FakeRunner(), root=PROJECT_ROOT, + ) + assert exit_code == 1 + assert "not found" in __import__("json").loads(output)["error"] \ No newline at end of file diff --git a/tools/elf-mcp/toolchain.py b/tools/elf-mcp/toolchain.py new file mode 100644 index 0000000..7f14320 --- /dev/null +++ b/tools/elf-mcp/toolchain.py @@ -0,0 +1,122 @@ +"""Toolchain resolution for the ARM GNU binutils used by elf-mcp. + +Resolves the ``arm-none-eabi-{size,nm,readelf,objdump,addr2line}`` executables +by searching (in order): + +1. Install trees under ``/Applications/ArmGNUToolchain/**/arm-none-eabi/bin`` + (newest version wins), +2. ``/opt/homebrew/bin`` and ``/usr/local/bin``, +3. ``$PATH`` (``shutil.which``). + +A ``binutils`` runner is provided so tools can be unit tested with a fake +executable resolver. +""" + +from __future__ import annotations + +import glob +import shutil +import subprocess +from dataclasses import dataclass +from typing import Callable, Protocol + +#: Tool names we are interested in. +TOOL_NAMES = ("size", "nm", "readelf", "objdump", "addr2line") + +#: Candidate install roots on this host (Darwin oriented, Harmless if absent). +CANDIDATE_ROOTS = ( + "/Applications/ArmGNUToolchain", + "/Applications/arm-gnu-toolchain", +) + + +def _arm_install_bin_dirs() -> list[str]: + """Return candidate ``arm-none-eabi/bin`` directories, newest release first.""" + dirs: list[str] = [] + for root in CANDIDATE_ROOTS: + for release_dir in glob.glob(f"{root}/*"): + bin_dir = f"{release_dir}/arm-none-eabi/bin" + if glob.glob(f"{bin_dir}/arm-none-eabi-size"): + dirs.append(bin_dir) + # Newest release first (versions like "14.2.rel1" sort lexicographically close + # enough for our purposes; a numeric sort below is more robust). + def _version_key(path: str) -> tuple[int, ...]: + tail = path.split("/")[-2] + nums = [] + for part in tail.split("."): + digits = "".join(ch for ch in part if ch.isdigit()) + try: + nums.append(int(digits)) + except ValueError: + nums.append(0) + return tuple(nums) + + dirs.sort(key=_version_key, reverse=True) + return dirs + + +def resolve_binutils_tool(tool: str, root: str | None = None) -> str | None: + """Return the absolute path to a binutils tool, or ``None`` if not found. + + ``tool`` is the short name without the ``arm-none-eabi-`` prefix (e.g. + ``"size"``). ``root`` optionally points at an explicit + ``.../arm-none-eabi/bin`` directory to skip all discovery. + """ + if tool not in TOOL_NAMES: + raise ValueError(f"unsupported binutils tool: {tool!r}") + if root: + cand = f"{root}/arm-none-eabi-{tool}" + return cand if glob.glob(cand) else None + for bin_dir in _arm_install_bin_dirs(): + cand = f"{bin_dir}/arm-none-eabi-{tool}" + if glob.glob(cand): + return cand + return shutil.which(f"arm-none-eabi-{tool}") + + +@dataclass +class CommandResult: + """Result of running a binutils command.""" + + returncode: int + stdout: str + stderr: str + + +class BinutilsRunner(Protocol): + """Minimal runner protocol; tools accept any callable matching it.""" + + def __call__(self, tool: str, args: list[str], *, cwd: str | None = None) -> CommandResult: + ... + + +def run_binutils(tool: str, args: list[str], *, root: str | None = None, + cwd: str | None = None, timeout: float = 60.0) -> CommandResult: + """Run ``arm-none-eabi- `` and return its captured output. + + Raises ``FileNotFoundError`` if the tool cannot be resolved. + """ + exe = resolve_binutils_tool(tool, root=root) + if exe is None: + raise FileNotFoundError( + f"arm-none-eabi-{tool} not found in any known location; install the ARM " + "GNU toolchain or add it to $PATH" + ) + proc = subprocess.run( + [exe, *args], + capture_output=True, + text=True, + cwd=cwd, + timeout=timeout, + check=False, + ) + return CommandResult(proc.returncode, proc.stdout or "", proc.stderr or "") + + +def make_runner(root: str | None = None) -> BinutilsRunner: + """Factory for a runner bound to a specific toolchain root (or auto-detected).""" + + def runner(tool: str, args: list[str], *, cwd: str | None = None) -> CommandResult: + return run_binutils(tool, args, root=root, cwd=cwd) + + return runner \ No newline at end of file diff --git a/tools/elf-mcp/tools.py b/tools/elf-mcp/tools.py new file mode 100644 index 0000000..4c303d1 --- /dev/null +++ b/tools/elf-mcp/tools.py @@ -0,0 +1,625 @@ +"""Tool implementations for the elf-mcp server. + +Every tool is a plain ``callable(params, runner) -> (exit_code, output_text)``. +``runner`` must match :class:`toolchain.BinutilsRunner` (defaults to the real +binutils via :func:`toolchain.make_runner`), so the tools are unit-testable +with a fake runner and the same entrypoints can be driven from a CLI or the +MCP server. +""" + +from __future__ import annotations + +import json +import os +import re +from typing import Callable + +from toolchain import BinutilsRunner, make_runner, run_binutils +from parsers import ( + MemoryRegion, + compute_size_breakdown, + parse_linker_memory, + parse_map_discarded, + parse_map_memory_config, + parse_map_per_object, + parse_nm_symbols, + parse_objdump_sections, + parse_readelf_sections, + region_for_address, +) + +PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) + +#: Sections that must never appear in the on-target image. Matched against +#: linker-mangled names and hard-coded symbol spellings. +HEAP_SYMBOLS = ( + "malloc", "calloc", "realloc", "free", + "_Znwm", "_Znam", "_Znaj", "_Znwj", # operator new / new[] + "_ZdlPv", "_ZdaPv", "_ZdlPvj", "_ZdaPvj", # operator delete / delete[] + "__cxa_throw", "__cxa_allocate_exception", "__cxa_free_exception", + "__cxa_rethrow", "_Unwind_", "__gxx_personality", + "_GLOBAL__sub_I", "__dso_handle", "_initterm", +) + + +def _resolve_path(path: str, root: str) -> str: + """Resolve a possibly-relative path against ``root`` (the project dir).""" + if os.path.isabs(path): + return path + return os.path.abspath(os.path.join(root, path)) + + +def _require_elf(params: dict, root: str) -> str: + elf = params.get("elf") + if not elf: + raise ValueError("'elf' argument is required") + elf = _resolve_path(elf, root) + if not os.path.isfile(elf): + raise FileNotFoundError(f"ELF file not found: {elf}") + return elf + + +def _sibling_map_path(elf: str) -> str: + return elf.rsplit(".", 1)[0] + ".map" + + +def _find_linkerscript(elf: str, root: str) -> str | None: + """Best-effort discovery of the linker script for a firmware ELF. + + Searches, in order: an explicit ``ld`` argument, a ``.map`` sibling (for the + ``Memory Configuration`` table), and the conventional ``linkerscripts/`` + layout under ``projects/``, ``modules/`` and ``boards/``. + """ + map_path = _sibling_map_path(elf) + if os.path.isfile(map_path): + return map_path + return None + + +def _json(output) -> tuple[int, str]: + return 0, json.dumps(output, indent=2) + + +def _err(msg: str) -> tuple[int, str]: + return 1, json.dumps({"error": msg}, indent=2) + + +def _regions_from_ld(ld_path: str) -> list[MemoryRegion]: + with open(ld_path) as handle: + return parse_linker_memory(handle.read()) + + +def _regions_from_map(map_path: str) -> list[MemoryRegion]: + with open(map_path) as handle: + return parse_map_memory_config(handle.read()) + + +# --------------------------------------------------------------------------- +# discovery +# --------------------------------------------------------------------------- + + +def list_elf_files(params: dict, runner: BinutilsRunner | None = None, + root: str = PROJECT_ROOT) -> tuple[int, str]: + """Discover firmware ``.elf`` / ``.map`` artifacts under a build root.""" + build_root = params.get("build_root", "build") + preset_filter = params.get("preset") or None + build_root = _resolve_path(build_root, root) + if not os.path.isdir(build_root): + return _err(f"build root not found: {build_root}") + artifacts = [] + for dirpath, dirnames, filenames in os.walk(build_root): + if ".dot." in dirpath: # graphviz scratch outputs + continue + if "native" in params.get("preset", "") or "host" in params.get("preset", ""): + pass + for fname in filenames: + if not fname.endswith(".elf"): + continue + if ".dot." in fname: # graphviz copy inside build dir + continue + rel = os.path.relpath(os.path.join(dirpath, fname), build_root) + preset = rel.split(os.sep)[0] + if preset_filter and not ( + preset == preset_filter + or preset_filter == f"on-target-{preset}" + or preset_filter.endswith(preset) + ): + continue + elf_path = os.path.join(dirpath, fname) + map_path = _sibling_map_path(elf_path) + artifacts.append({ + "preset": preset, + "relative": rel, + "elf": elf_path, + "map": map_path if os.path.isfile(map_path) else None, + "modified": os.path.getmtime(elf_path), + }) + artifacts.sort(key=lambda a: a["relative"]) + return _json({"count": len(artifacts), "artifacts": artifacts}) + + +# --------------------------------------------------------------------------- +# memory / section analysis +# --------------------------------------------------------------------------- + + +def binary_size(params: dict, runner: BinutilsRunner | None = None, + root: str = PROJECT_ROOT) -> tuple[int, str]: + """Flash/RAM totals and per-section classification from an ELF.""" + runner = runner or make_runner() + elf = _require_elf(params, root) + result = runner("objdump", ["-h", elf]) + if result.returncode != 0: + return _err(f"objdump failed: {result.stderr.strip()}") + sections = parse_objdump_sections(result.stdout) + breakdown = compute_size_breakdown(sections) + flash_rows = [] + for name, fbytes, rbytes in breakdown.sections: + flash_rows.append({ + "section": name, + "flash": fbytes, + "ram": rbytes, + }) + return _json({ + "elf": elf, + "flash": breakdown.flash, + "ram": breakdown.ram, + "total": breakdown.total, + "sections": flash_rows, + "note": ".data-like sections are charged to flash (LMA copy) and RAM (VMA); " + "stack && DMA reservations are linker-reserved RAM.", + }) + + +def section_breakdown(params: dict, runner: BinutilsRunner | None = None, + root: str = PROJECT_ROOT) -> tuple[int, str]: + """Verbose ``readelf -S``-derived listing of every section.""" + runner = runner or make_runner() + elf = _require_elf(params, root) + result = runner("readelf", ["-SW", elf]) + if result.returncode != 0: + return _err(f"readelf failed: {result.stderr.strip()}") + sections = parse_readelf_sections(result.stdout) + rows = [ + { + "number": s.number, + "name": s.name, + "type": s.type, + "address": s.addr, + "offset": s.offset, + "size": s.size, + "flags": s.flags, + } + for s in sections + ] + return _json({"elf": elf, "count": len(rows), "sections": rows}) + + +def memory_map(params: dict, runner: BinutilsRunner | None = None, + root: str = PROJECT_ROOT) -> tuple[int, str]: + """Map ELF sections onto linker ``MEMORY`` regions with fill percentages.""" + runner = runner or make_runner() + elf = _require_elf(params, root) + ld_path = params.get("ld") + if ld_path: + ld_path = _resolve_path(ld_path, root) + regions = [] + if ld_path and os.path.isfile(ld_path): + regions = _regions_from_ld(ld_path) + else: + map_path = _sibling_map_path(elf) + if os.path.isfile(map_path): + regions = _regions_from_map(map_path) + if not regions: + return _err( + "no MEMORY regions found: pass an explicit 'ld' linkerscript path " + f"(checked {_sibling_map_path(elf)})" + ) + + result = runner("objdump", ["-h", elf]) + if result.returncode != 0: + return _err(f"objdump failed: {result.stderr.strip()}") + sections = parse_objdump_sections(result.stdout) + + region_sizes: dict[str, int] = {r.name: 0 for r in regions} + section_placement: list[dict] = [] + for section in sections: + region = region_for_address(regions, section.vma) or region_for_address( + regions, section.lma + ) + if region is None: + continue + region_sizes[region.name] += section.size + section_placement.append({ + "section": section.name, + "size": section.size, + "vma": section.vma, + "lma": section.lma, + "region": region.name, + "writable": region.is_writable, + }) + + region_rows = [] + for region in regions: + used = region_sizes.get(region.name, 0) + if used == 0: + continue + region_rows.append({ + "region": region.name, + "used": used, + "length": region.length, + "fill_percent": round(100.0 * used / region.length, 2), + "kind": "ram" if region.is_writable else "flash", + }) + region_rows.sort(key=lambda r: r["fill_percent"], reverse=True) + return _json({ + "elf": elf, + "linker_script": ld_path or _sibling_map_path(elf), + "regions": region_rows, + "section_placement": section_placement, + }) + + +def largest_symbols(params: dict, runner: BinutilsRunner | None = None, + root: str = PROJECT_ROOT) -> tuple[int, str]: + """Top-N symbols by size from ``nm`` (bloat hunting).""" + runner = runner or make_runner() + elf = _require_elf(params, root) + count = int(params.get("count", 25)) + result = runner("nm", ["-S", "--size-sort", "--radix=x", elf]) + if result.returncode != 0: + return _err(f"nm failed: {result.stderr.strip()}") + symbols = parse_nm_symbols(result.stdout) + sized = [s for s in symbols if s.size is not None and s.address is not None] + sized.sort(key=lambda s: (s.size or 0), reverse=True) + top = [] + for symbol in sized[:count]: + top.append({ + "name": symbol.name, + "address": symbol.address, + "size": symbol.size, + "kind": symbol.kind, + }) + total_text = sum(s.size or 0 for s in sized if s.kind == "text") + return _json({ + "elf": elf, + "count": len(top), + "total_text_bytes": total_text, + "symbols": top, + }) + + +def per_object_map(params: dict, runner: BinutilsRunner | None = None, + root: str = PROJECT_ROOT) -> tuple[int, str]: + """Per-object flash/RAM contributions from the linker map file.""" + runner = runner or make_runner() + elf = params.get("elf") + map_path = params.get("map") + if not elf and not map_path: + return _err("'elf' or 'map' argument is required") + if map_path: + map_path = _resolve_path(map_path, root) + else: + elf = _require_elf(params, root) + map_path = _sibling_map_path(elf) + if not os.path.isfile(map_path): + return _err(f"map file not found: {map_path} (build firmware with -Map)") + with open(map_path) as handle: + text = handle.read() + regions = parse_map_memory_config(text) + blames = parse_map_per_object(text, regions) + top = int(params.get("top", 15)) + rows = [ + { + "object": b.object, + "flash": b.flash, + "ram": b.ram, + "total": b.total, + } + for b in blames[:top] + ] + covered_flash = sum(b.flash for b in blames) + covered_ram = sum(b.ram for b in blames) + discarded = parse_map_discarded(text) + return _json({ + "elf": elf, + "map": map_path, + "objects": rows, + "covered_flash": covered_flash, + "covered_ram": covered_ram, + "discarded_sections": len(discarded), + "note": "Totals exclude linker-reserved regions (stacks, DMA buffers) " + "and .data LMA copies; use binary_size for exact image totals.", + }) + + +# --------------------------------------------------------------------------- +# code inspection +# --------------------------------------------------------------------------- + +#: Regular expressions that reliably name a heap-related routine in ARM +#: disassembly once demangled. Kept as a tuple for testability. +_HEAP_DISASSEMBLY_RE = re.compile( + r"\b(malloc|calloc|realloc|free|_Znwm|_Znam|_Znaj|_Znwj|_ZdlPv|_ZdaPv" + r"|__cxa_throw|__cxa_allocate_exception|__cxa_free_exception|__cxa_rethrow)\b" +) + + +def verify_no_heap(params: dict, runner: BinutilsRunner | None = None, + root: str = PROJECT_ROOT) -> tuple[int, str]: + """Scan the image for dynamic memory / exception machinery. + + Reports any heap-ish symbols found in the symbol table or as call targets in + the disassembly. A clean image exits 0 with ``violations: []``. + """ + runner = runner or make_runner() + elf = _require_elf(params, root) + include_unwind = bool(params.get("include_unwind", False)) + violations: list[dict] = [] + + nm_result = runner("nm", ["--defined-only", elf]) + if nm_result.returncode == 0: + for symbol in parse_nm_symbols(nm_result.stdout): + lowered = symbol.name + found = any( + needle in lowered + for needle in ("malloc", "calloc", "realloc") + ) or any( + needle in lowered + for needle in ("_Znwm", "_Znam", "_ZdlPv", "_ZdaPv", + "__cxa_", "_GLOBAL__sub_I", "__dso_handle") + ) + if include_unwind: + found = found or lowered.startswith("_Unwind_") or "__gxx_personality" in lowered + if found: + violations.append({ + "source": "symbol_table", + "symbol": symbol.name, + "address": symbol.address, + }) + + objdump_result = runner("objdump", ["-d", elf]) + if objdump_result.returncode == 0: + for lineno, line in enumerate(objdump_result.stdout.splitlines(), 1): + if _HEAP_DISASSEMBLY_RE.search(line): + violations.append({ + "source": "disassembly", + "line": lineno, + "line_text": line.strip(), + }) + + pass_result = not violations + return _json({ + "elf": elf, + "pass": pass_result, + "violations": violations[: int(params.get("limit", 40))], + "checked_symbols": list(HEAP_SYMBOLS), + }) + + +def disassemble(params: dict, runner: BinutilsRunner | None = None, + root: str = PROJECT_ROOT) -> tuple[int, str]: + """Disassemble one function/address range from the ELF.""" + runner = runner or make_runner() + elf = _require_elf(params, root) + symbol = params.get("symbol") + address = params.get("address") + if not symbol and not address: + return _err("'symbol' or 'address' argument is required") + if symbol: + nm_result = runner("nm", ["-S", elf]) + if nm_result.returncode != 0: + return _err(f"nm failed: {nm_result.stderr.strip()}") + matches = [ + (s.address, s.size) + for s in parse_nm_symbols(nm_result.stdout) + if s.name == symbol and s.address is not None and s.size is not None + ] + if not matches: + return _err(f"symbol not found in ELF: {symbol}") + start, size = matches[0] + if size == 0: + size = 0x100 + start_arg = hex(start) + stop_arg = hex(start + size) + desc = {"symbol": symbol, "start": start, "size": size} + else: + if not isinstance(address, str): + return _err("'address' must be a hex string") + start = int(address, 0) + span = int(params.get("span", 0x80)) + start_arg = hex(start) + stop_arg = hex(start + span) + desc = {"address": start, "span": span} + result = runner( + "objdump", + ["-d", "--no-show-raw-insn", f"--start-address={start_arg}", + f"--stop-address={stop_arg}", elf], + ) + if result.returncode != 0: + return _err(f"objdump failed: {result.stderr.strip()}") + return _json({"elf": elf, **desc, "disassembly": result.stdout}) + + +def addr2line(params: dict, runner: BinutilsRunner | None = None, + root: str = PROJECT_ROOT) -> tuple[int, str]: + """Resolve one or more addresses to function + source line (DWARF).""" + runner = runner or make_runner() + elf = _require_elf(params, root) + raw = params.get("address") + if not raw: + return _err("'address' argument is required") + addresses = raw if isinstance(raw, list) else [raw] + resolved = [] + for address in addresses: + result = runner("addr2line", ["-e", elf, "-f", "-i", str(address)]) + lines = [ln for ln in result.stdout.splitlines() if ln.strip()] + resolved.append({ + "address": address, + "function": lines[0] if lines else "??", + "source": lines[-1] if len(lines) > 1 else "??", + }) + return _json({"elf": elf, "resolved": resolved}) + + +#: Registry consumed by the MCP server. Keeping it here lets the CLI and +#: server share the same tool surface. +TOOLS: list[dict] = [ + { + "name": "list_elf_files", + "description": "Discover firmware .elf/.map artifacts under a CMake build root. " + "Returns the preset, relative path, and map sibling for each ELF.", + "inputSchema": { + "type": "object", + "properties": { + "build_root": {"type": "string", "description": "Build directory (default 'build')"}, + "preset": {"type": "string", "description": "Optional preset name filter"}, + }, + }, + "handler": list_elf_files, + }, + { + "name": "binary_size", + "description": "Flash/RAM totals for an ELF with per-section classification. " + ".data-like sections are charged to flash (LMA copy) and RAM (VMA).", + "inputSchema": { + "type": "object", + "properties": { + "elf": {"type": "string", "description": "Path to firmware ELF (absolute or relative)"}, + }, + "required": ["elf"], + }, + "handler": binary_size, + }, + { + "name": "section_breakdown", + "description": "Verbose per-section listing (readelf -S) including debug sections. " + "Use for address/offset/flags inspection.", + "inputSchema": { + "type": "object", + "properties": { + "elf": {"type": "string", "description": "Path to firmware ELF"}, + }, + "required": ["elf"], + }, + "handler": section_breakdown, + }, + { + "name": "memory_map", + "description": "Fill percentage of each linker MEMORY region by the ELF's sections. " + "Resolves regions from the .map sibling, or from an explicit linker script.", + "inputSchema": { + "type": "object", + "properties": { + "elf": {"type": "string", "description": "Path to firmware ELF"}, + "ld": {"type": "string", "description": "Optional linker script path"}, + }, + "required": ["elf"], + }, + "handler": memory_map, + }, + { + "name": "largest_symbols", + "description": "Top-N symbols by size from nm (flash/RAM bloat hunting).", + "inputSchema": { + "type": "object", + "properties": { + "elf": {"type": "string", "description": "Path to firmware ELF"}, + "count": {"type": "integer", "description": "How many symbols (default 25)"}, + }, + "required": ["elf"], + }, + "handler": largest_symbols, + }, + { + "name": "per_object_map", + "description": "Per-object flash/RAM contributions parsed from the linker .map file. " + "Takes 'elf' (map sibling guessed) or an explicit 'map' path.", + "inputSchema": { + "type": "object", + "properties": { + "elf": {"type": "string", "description": "Firmware ELF (map sibling derived)"}, + "map": {"type": "string", "description": "Explicit .map path"}, + "top": {"type": "integer", "description": "Objects to list (default 15)"}, + }, + }, + "handler": per_object_map, + }, + { + "name": "verify_no_heap", + "description": "Scan the image for dynamic-memory/exception machinery " + "(malloc, operator new, __cxa_*, static init). Pass/fail + violations.", + "inputSchema": { + "type": "object", + "properties": { + "elf": {"type": "string", "description": "Path to firmware ELF"}, + "include_unwind": {"type": "boolean", + "description": "Also flag _Unwind_/personality routines"}, + "limit": {"type": "integer", "description": "Max violations to report (default 40)"}, + }, + "required": ["elf"], + }, + "handler": verify_no_heap, + }, + { + "name": "disassemble", + "description": "Disassemble one function (by symbol) or an address range.", + "inputSchema": { + "type": "object", + "properties": { + "elf": {"type": "string", "description": "Path to firmware ELF"}, + "symbol": {"type": "string", "description": "Function name to disassemble"}, + "address": {"type": "string", "description": "Start address (hex), used with span"}, + "span": {"type": "integer", "description": "Bytes after address (default 0x80)"}, + }, + "required": ["elf"], + }, + "handler": disassemble, + }, + { + "name": "addr2line", + "description": "Resolve address(es) to function + source file:line via DWARF. " + "Pairs with pylink-square-mcp backtraces.", + "inputSchema": { + "type": "object", + "properties": { + "elf": {"type": "string", "description": "Path to firmware ELF"}, + "address": { + "oneOf": [ + {"type": "string", "description": "Single hex address"}, + {"type": "array", "items": {"type": "string"}}, + ], + "description": "One or more addresses", + }, + }, + "required": ["elf", "address"], + }, + "handler": addr2line, + }, +] + + +def main() -> None: + """CLI entrypoint: ``python tools.py --json '{...}'``.""" + import argparse + import sys + + parser = argparse.ArgumentParser(description="elf-mcp tool CLI") + parser.add_argument("tool", choices=[t["name"] for t in TOOLS]) + parser.add_argument("--json", required=True, help="JSON arguments object") + args = parser.parse_args() + + tool = next(t for t in TOOLS if t["name"] == args.tool) + try: + params = json.loads(args.json) + except json.JSONDecodeError as exc: + print(json.dumps({"error": f"invalid JSON: {exc}"})) + sys.exit(1) + exit_code, output = tool["handler"](params) + print(output) + sys.exit(exit_code) + + +if __name__ == "__main__": + main() \ No newline at end of file