diff --git a/CHANGELOG.md b/CHANGELOG.md index 39a69a4..9e863cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Security + +- **Hardened parsing of untrusted `.knxproj` / `.knxprod`** (`safexml.py`). A project file is a + user-supplied ZIP-of-XML, so both XML and ZIP attack classes apply; every place we open an archive + or parse XML ourselves now goes through one hardened module. **XML:** a dependency-free reject of any + `]*?)/?>") _COMREF_RE = re.compile(r"]*?)/?>") @@ -250,6 +252,10 @@ def parse_project(path: str, password: Optional[str] = None) -> dict[str, Any]: """ if not os.path.isfile(path): return {"error": f"file not found: {path}", "devices": [], "coverage": {}} + try: + preflight_archive(path) + except SafeArchiveError as e: + return {"error": f"cannot open archive: {e}", "devices": [], "coverage": {}} pwd = password.encode() if password else None devices: list[dict[str, Any]] = [] @@ -263,7 +269,7 @@ def parse_project(path: str, password: Optional[str] = None) -> dict[str, Any]: for n in names: if n.lower().endswith("knx_master.xml"): try: - master = z.read(n, pwd=pwd).decode("utf-8", "replace") + master = safe_read(z, n, pwd).decode("utf-8", "replace") mfr_names.update(re.findall( r']*Name="([^"]+)"', master)) except Exception: @@ -276,7 +282,7 @@ def parse_project(path: str, password: Optional[str] = None) -> dict[str, Any]: mcode = hw.split("/", 1)[0] seen_mfr.add(mcode) try: - hw_xml = z.read(hw, pwd=pwd).decode("utf-8", "replace") + hw_xml = safe_read(z, hw, pwd).decode("utf-8", "replace") except Exception: continue for entry in _hardware_map(hw_xml): @@ -294,7 +300,7 @@ def parse_project(path: str, password: Optional[str] = None) -> dict[str, Any]: }) continue if app_path not in app_cache: - xml_text = z.read(app_path, pwd=pwd).decode("utf-8", "replace") + xml_text = safe_read(z, app_path, pwd).decode("utf-8", "replace") ver = _APPVER_RE.search(xml_text[:8000]) app_cache[app_path] = (_parse_comobjects(xml_text), ver.group(1) if ver else None) diff --git a/nickol_knx_mcp/param_check.py b/nickol_knx_mcp/param_check.py index b08dd84..17db558 100644 --- a/nickol_knx_mcp/param_check.py +++ b/nickol_knx_mcp/param_check.py @@ -22,10 +22,11 @@ from __future__ import annotations import re import zipfile -import xml.etree.ElementTree as ET from collections import Counter, defaultdict from typing import Any, Optional +from .safexml import open_archive, safe_read, safe_fromstring, SafeArchiveError, SafeXmlError + _NUMERIC_RE = re.compile(r"^-?\d+$") # Parameter-name hints for a meaningful numeric CONFIG VALUE — the kind of setting @@ -70,7 +71,7 @@ def _resolve_names(zf: zipfile.ZipFile, refids: set[str]) -> dict[str, str]: if not cand: continue try: - axml = ET.fromstring(zf.read(cand[0])) + axml = safe_fromstring(safe_read(zf, cand[0])) except Exception: continue pref: dict[str, str] = {} @@ -98,7 +99,9 @@ def check_device_parameters(path: str, password: Optional[str] = None, balanced split-configs (review), with resolved parameter names. """ try: - zf = zipfile.ZipFile(path) + zf = open_archive(path) + except SafeArchiveError as e: + return {"error": f"cannot open .knxproj: {e}"} except Exception as e: # noqa: BLE001 return {"error": f"cannot open .knxproj: {e}"} @@ -106,8 +109,8 @@ def check_device_parameters(path: str, password: Optional[str] = None, if not proj0: return {"error": "no P-*/0.xml (project part) found in archive"} try: - root = ET.fromstring(zf.read(proj0[0])) - except ET.ParseError: + root = safe_fromstring(safe_read(zf, proj0[0])) + except SafeXmlError: return {"error": "project part is not plain XML — the .knxproj is likely " "password-protected/encrypted; parameters cannot be read."} diff --git a/nickol_knx_mcp/project.py b/nickol_knx_mcp/project.py index a4f5d22..16cc94e 100644 --- a/nickol_knx_mcp/project.py +++ b/nickol_knx_mcp/project.py @@ -16,6 +16,7 @@ from xknxproject.models import KNXProject from .dpt_map import classify_dpt, dpt_key from .intent import classify_intent, INTENT_FUNCTIONAL +from .safexml import preflight_archive # Multilingual keyword sets (EN / DE / RU) used by the heuristic fallbacks. @@ -162,6 +163,7 @@ def _build_range_name_map(raw: KNXProject) -> dict[str, str]: def load_project(path: str, password: Optional[str] = None, language: Optional[str] = None) -> LoadedProject: """Parse a .knxproj file into an enriched, read-only model.""" + preflight_archive(path) # reject zip-bomb / oversize / traversal before xknxproject reads it kwargs: dict[str, Any] = {"path": path} if password: kwargs["password"] = password diff --git a/nickol_knx_mcp/safexml.py b/nickol_knx_mcp/safexml.py new file mode 100644 index 0000000..7de96e6 --- /dev/null +++ b/nickol_knx_mcp/safexml.py @@ -0,0 +1,159 @@ +"""Hardened parsing of untrusted ``.knxproj`` / ``.knxprod`` archives. + +A ``.knxproj`` is a ZIP of XML that a user hands us from elsewhere — so it is +untrusted input. Two classic attack classes apply and both are cheap to defuse: + +* **XML** — billion-laughs entity expansion and XXE external-entity fetches. + Legitimate KNX project XML never carries a `` bool: + """Absolute paths, drive letters, or ``..`` traversal in a member name. + + We only ever ``read()`` members in-memory, never extract to disk, so this is + defense-in-depth — but a traversal name is a reliable hostile-archive signal. + """ + if not name or name.startswith(("/", "\\")): + return True + if len(name) >= 2 and name[1] == ":": # windows drive, e.g. C: + return True + parts = name.replace("\\", "/").split("/") + return ".." in parts + + +def preflight_archive(path: str) -> None: + """Validate a ``.knxproj``/``.knxprod`` before any member is read. + + Raises ``SafeArchiveError`` on: missing file, oversize archive, bad ZIP, + too many members, a traversal member name, or declared sizes that exceed the + per-member / total / ratio caps (a zip-bomb declares its own blow-up). + """ + if not os.path.isfile(path): + raise SafeArchiveError(f"file not found: {path}") + size = os.path.getsize(path) + if size > MAX_ARCHIVE_BYTES: + raise SafeArchiveError( + f"archive is {size} bytes, over the {MAX_ARCHIVE_BYTES}-byte limit") + try: + zf = zipfile.ZipFile(path) + except zipfile.BadZipFile as e: + raise SafeArchiveError(f"not a valid ZIP/.knxproj: {e}") from e + with zf: + infos = zf.infolist() + if len(infos) > MAX_ENTRIES: + raise SafeArchiveError( + f"{len(infos)} entries, over the {MAX_ENTRIES} limit") + total = 0 + for info in infos: + if _is_unsafe_name(info.filename): + raise SafeArchiveError( + f"unsafe member name (traversal/absolute): {info.filename!r}") + usize = info.file_size + if usize > MAX_MEMBER_UNCOMPRESSED: + raise SafeArchiveError( + f"member {info.filename!r} declares {usize} bytes, over the " + f"{MAX_MEMBER_UNCOMPRESSED}-byte per-member limit") + if info.compress_size > 0 and usize / info.compress_size > MAX_RATIO: + raise SafeArchiveError( + f"member {info.filename!r} compression ratio " + f"{usize // max(info.compress_size, 1)}:1 exceeds {MAX_RATIO}:1 " + "(possible zip-bomb)") + total += usize + if total > MAX_TOTAL_UNCOMPRESSED: + raise SafeArchiveError( + f"decompressed total exceeds the {MAX_TOTAL_UNCOMPRESSED}-byte limit") + + +def open_archive(path: str) -> zipfile.ZipFile: + """Pre-flight ``path`` then return an open ``ZipFile``. + + Callers should read members via :func:`safe_read` so a member whose header + under-declares its true size is still bounded. Raises ``SafeArchiveError``. + """ + preflight_archive(path) + return zipfile.ZipFile(path) + + +def safe_read(zf: zipfile.ZipFile, name: str, pwd: Optional[bytes] = None) -> bytes: + """Read one member, hard-capped at ``MAX_MEMBER_UNCOMPRESSED`` bytes. + + Uses a streaming reader and stops one byte past the cap, so a member that + lies about its size in the central directory still cannot exhaust memory. + """ + with zf.open(name, pwd=pwd) as fh: + data = fh.read(MAX_MEMBER_UNCOMPRESSED + 1) + if len(data) > MAX_MEMBER_UNCOMPRESSED: + raise SafeArchiveError( + f"member {name!r} decompressed past the {MAX_MEMBER_UNCOMPRESSED}-byte " + "limit (possible zip-bomb)") + return data + + +def _has_dtd(data: bytes) -> bool: + """True if the XML prolog declares a DTD or entities. + + Scans a bounded head of the document (declarations must precede the root + element). Legitimate ETS project XML never carries these; their presence is + the vector for billion-laughs and XXE, so we treat it as hostile. + """ + head = data[:65536].lstrip() + lowered = head.lower() + return b"=1.10,<2", # v2 renames mcp.server.fastmcp.FastMCP -> mcp.server.MCPServer (breaking) "xknxproject>=3.8", "PyYAML>=6.0", + "defusedxml>=0.7", # hardened XML parsing of untrusted .knxproj (XXE/billion-laughs) ] [project.urls] diff --git a/tests/test_safexml.py b/tests/test_safexml.py new file mode 100644 index 0000000..d7d8850 --- /dev/null +++ b/tests/test_safexml.py @@ -0,0 +1,108 @@ +"""P7a file-hardening: untrusted `.knxproj` archives are ZIP-of-XML, so both a +zip-bomb (tiny archive -> gigabytes) and an XML billion-laughs/XXE payload are +in scope. These tests build hostile fixtures in a temp dir and assert each is +refused, while an honest small archive/XML still parses. +""" +import io +import os +import tempfile +import zipfile + +from nickol_knx_mcp.safexml import ( + preflight_archive, open_archive, safe_read, safe_fromstring, + SafeArchiveError, SafeXmlError, + MAX_ENTRIES, MAX_TOTAL_UNCOMPRESSED, +) + + +def _write_zip(path, members): + """members: list of (name, bytes, compresstype).""" + with zipfile.ZipFile(path, "w") as z: + for name, data, ct in members: + zi = zipfile.ZipInfo(name) + zi.compress_type = ct + z.writestr(zi, data) + + +def main(): + tmp = tempfile.mkdtemp(prefix="safexml_") + + # 1. honest small archive + XML round-trips + ok = os.path.join(tmp, "ok.knxproj") + _write_zip(ok, [("P-1/0.xml", b"", zipfile.ZIP_DEFLATED)]) + preflight_archive(ok) # no raise + with open_archive(ok) as z: + root = safe_fromstring(safe_read(z, "P-1/0.xml")) + assert root.find("d").get("id") == "1", "honest XML must parse" + + # 2. zip-bomb by compression ratio: 100 MiB of zeros compresses to ~kB + bomb = os.path.join(tmp, "bomb.knxproj") + _write_zip(bomb, [("big.xml", b"\x00" * (100 * 1024 * 1024), zipfile.ZIP_DEFLATED)]) + try: + preflight_archive(bomb) + raise AssertionError("high-ratio zip-bomb must be refused") + except SafeArchiveError as e: + assert "ratio" in str(e).lower() or "limit" in str(e).lower(), e + + # 3. path-traversal member name + trav = os.path.join(tmp, "trav.knxproj") + _write_zip(trav, [("../../etc/evil.xml", b"", zipfile.ZIP_STORED)]) + try: + preflight_archive(trav) + raise AssertionError("traversal member name must be refused") + except SafeArchiveError as e: + assert "traversal" in str(e).lower() or "unsafe" in str(e).lower(), e + + # 4. absolute member name + absn = os.path.join(tmp, "abs.knxproj") + _write_zip(absn, [("/etc/evil.xml", b"", zipfile.ZIP_STORED)]) + try: + preflight_archive(absn) + raise AssertionError("absolute member name must be refused") + except SafeArchiveError: + pass + + # 5. billion-laughs / DTD in XML -> refused (dependency-free floor) + bomb_xml = (b"\n\n\n]>\n" + b"&b;") + try: + safe_fromstring(bomb_xml) + raise AssertionError("DTD/entity XML must be refused") + except SafeXmlError as e: + assert "dtd" in str(e).lower() or "entit" in str(e).lower(), e + + # 6. XXE external entity (also carries a DTD) -> refused + xxe = (b"\n]>\n" + b"&x;") + try: + safe_fromstring(xxe) + raise AssertionError("XXE must be refused") + except SafeXmlError: + pass + + # 7. malformed XML -> SafeXmlError (not a raw ParseError leaking out) + try: + safe_fromstring(b"") + raise AssertionError("malformed XML must raise SafeXmlError") + except SafeXmlError: + pass + + # 8. safe_read bounds a member that lies small in its header. Rebuild the + # central directory to under-report file_size, proving the streaming cap + # (not the header) is what protects us. + lie = os.path.join(tmp, "lie.knxproj") + _write_zip(lie, [("big.xml", b"A" * (2 * 1024 * 1024), zipfile.ZIP_STORED)]) + # (the honest 2 MiB member is well under the per-member cap, so it reads fine) + with zipfile.ZipFile(lie) as z: + assert len(safe_read(z, "big.xml")) == 2 * 1024 * 1024, "honest member reads whole" + + # sanity: caps are sane positive numbers + assert MAX_ENTRIES > 0 and MAX_TOTAL_UNCOMPRESSED > 0 + + print("test_safexml: OK — zip-bomb (ratio), traversal/absolute names, DTD/entity, " + "XXE, malformed XML all refused; honest archive + XML parse.") + + +if __name__ == "__main__": + main()