github · esp32_nat_routerHEAD

Evidence

#!/usr/bin/env python3

"""

ESP32 NAT Router MCP Bridge

MCP server that controls an ESP32 NAT Router via its remote console (telnet/TCP).

Uses fastmcp and telnetlib3.

Usage:

    python esp_nat_bridge.py                          # stdio mode (for MCP clients)

    python esp_nat_bridge.py --transport streamable-http --port 8000  # HTTP mode

Environment variables:

    ESP_NAT_HOST     - Router IP address (default: 192.168.4.1)

    ESP_NAT_PORT     - Remote console TCP port (default: 2323)

    ESP

Remediation

Missing annotation: add `{ annotations: { destructiveHint: true } }` as the 4th argument to `server.tool(...)` (TS) or pass `destructive_hint=True` to the `@mcp.tool(...)` decorator (Python). readOnly-lying: remove the `readOnlyHint: true` / `read_only_hint=True` assertion, or refactor the handler so it no longer calls the destructive sink. Do not suppress with a waiver — clients rely on these hints for user-facing confirmation.

Evidence

document.getElementById('importStatus').textContent='Uploading...';\

var hdrs={'Content-Type':'application/json'};\

if(pass)hdrs['X-Config-Pass']=pass;\

fetch('/api/config-import',{method:'POST',body:r.result,headers:hdrs})\

.then(function(res){return res.json();})\

.then(function(d){\

if(d.ok){\

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

function downloadConfig(){\

var pass=document.getElementById('expPass').value;\

document.getElementById('exportStatus').textContent='Downloading...';\

fetch('/api/config-export',{method:'POST',body:JSON.stringify({pass:pass}),headers:{'Content-Type':'application/json'}})\

.then(function(res){\

if(!res.ok){return res.text().then(function(t){throw new Error(t);});}\

var cd=res.headers.get('Content-Disposition')||'';\

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

if self.writer:

            try:

                self.writer.write("quit\r\n")

                self.writer.close()

            except Exception:

                pass

        self.reader = None

        self.writer = None

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

while True:

                    chunk = await asyncio.wait_for(conn.reader.read(4096), timeout=0.2)

                    if not chunk:

                        break

            except asyncio.TimeoutError:

                pass

            conn.writer.write(f"ping {host}\r\n")

            response = await conn._read_until_prompt(timeout=15)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

finally:

            try:

                os.unlink(tmp.name)

            except OSError:

                pass

    return "\n".join(result_parts)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

self.reader.read(4096), timeout=0.2

                        )

                        if not chunk:

                            break

                except asyncio.TimeoutError:

                    pass

                self.writer.write(command + "\r\n")

                response = await self._read_until_prompt()

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

buf += chunk

                # The ESP32 console prompt is typically "esp32> " or just "> "

                if "esp32>" in buf or "\n> " in buf or buf.rstrip().endswith(">"):

                    break

        except asyncio.TimeoutError:

            pass

        return buf

    async def send_command(self, command: str) -> str:

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

finally:

        writer.close()

        try:

            await writer.wait_closed()

        except Exception:

            pass

    return bytes(buf)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

finally:

        try:

            os.unlink(tmp.name)

        except OSError:

            pass

# ═══════════════════════════════════════════════════════════════════════════

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.