diff --git a/CHANGELOG.md b/CHANGELOG.md index 458a322..7ec0a67 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.4.0] — 2026-07-01 + +**Two new lint dimensions: does the DPT sub-type match what the name promises, and is the +project's KNX Secure posture consistent.** This release adds a conservative sub-DPT sanity +linter and a report-only KNX Data Secure posture summary (no key material ever touched), +plus the itemised QA findings in the handover pack. + ### Added +- **A1 — sub-DPT sanity linter** (`analyze.py`, surfaced by `check_dpt` and `analyze_all` as the + `subdpt_suspect` finding). When a group-address name implies a specific DPT sub-type + (temperature → `9.001`, power → `14.056`, brightness/position → `5.001`, and similar), the + linter flags a wrong sub-type or a wrong main type. Multilingual keyword matching, deliberately + conservative — it only fires when the name is unambiguous, so it does not second-guess correct + or generic DPTs. +- **A4 — KNX Data Secure posture** (`analyze.py` `secure_posture()`, new MCP tool `check_secure`). + A report-only summary of the project's security posture: secured vs plaintext GA counts, middle + groups that mix secure and plaintext objects, and a keyring (`.knxkeys`) handover checklist. It + reads only the per-GA `Security` flag — no key material is read, derived, or emitted. +- **KNX Secure posture section in the handover pack** — section 5 of `handover.md` is rewritten + from a flat secure-GA count into the full posture section (counts, mixed-group flag, keyring + handover checklist), so the as-built deliverable states the security posture explicitly. - **Itemised QA findings in the handover pack** — section 6 of `handover.md` now lists the actual 🔴 errors and 🟡 warnings (address + name, grouped by check), not just totals, so the handover doubles as a review checklist. Info-level items (intentional reserves / logic / macros) diff --git a/nickol_knx_mcp/__init__.py b/nickol_knx_mcp/__init__.py index 420ce24..98c485f 100644 --- a/nickol_knx_mcp/__init__.py +++ b/nickol_knx_mcp/__init__.py @@ -1,2 +1,2 @@ """nickol-knx-mcp: design-time KNX/ETS project assistant MCP server.""" -__version__ = "0.3.0" +__version__ = "0.4.0" diff --git a/nickol_knx_mcp/analyze.py b/nickol_knx_mcp/analyze.py index e80f685..6615f05 100644 --- a/nickol_knx_mcp/analyze.py +++ b/nickol_knx_mcp/analyze.py @@ -170,6 +170,33 @@ def _is_central_macro(name: str) -> bool: return any(t in low for t in _CENTRAL_MACRO_TOKENS) +# --------------------------------------------------------------------------- # +# Sub-DPT sanity (A1): a name implies a specific DPT sub-type. Flag when the +# main matches but the sub is wrong (9.001 temp vs 9.004 lux), or — for strong +# physical quantities — when the main itself is wrong for the named function. +# Multilingual (RU / EN / DE). Conservative: only fires on clear function tokens. +# (tokens, main, sub, strong) — strong => also flag a wrong main. +# --------------------------------------------------------------------------- # +_SUBDPT_RULES: tuple = ( + (("влажност", "humidity", "feucht"), 9, 7, True), + (("co2", "со2", "углекисл", "kohlendioxid"), 9, 8, True), + (("освещённост", "освещенност", "luminosity", "lux", "helligkeit ("), 9, 4, True), + (("температур", "temperatur"), 9, 1, True), + (("мощност", "power ", "leistung"), 14, 56, True), + (("энерги", "energy", "energie", "квтч", "kwh"), 13, 13, True), + (("яркост", "brightness", "значение яркости", "dimmwert", "helligkeitswert"), 5, 1, False), + (("позици", "position", "stellung"), 5, 1, False), +) + + +def _expected_subdpt(name: str) -> Optional[tuple[int, int, bool]]: + low = (name or "").lower() + for tokens, m, s, strong in _SUBDPT_RULES: + if any(t in low for t in tokens): + return (m, s, strong) + return None + + # command DPT main -> acceptable status DPT mains _STATUS_COMPAT = { 1: {1}, # switch command -> 1.x status @@ -310,4 +337,82 @@ def detect_dpt_issues(project: LoadedProject) -> list[dict[str, Any]]: addresses=[r.address for r in recs], dpts=sorted(dpts), )) + # 4. sub-DPT sanity — the name implies a specific sub-type (A1) + for addr, ga in project.gas.items(): + if ga.intent != INTENT_FUNCTIONAL or ga.dpt_main is None: + continue + exp = _expected_subdpt(ga.name) + if exp is None: + continue + em, es, strong = exp + if ga.dpt_main == em and ga.dpt_sub != es: + findings.append(_finding( + SEVERITY_WARN, "subdpt_suspect", addr, + f"'{ga.name}' looks like a {em}.{es:03d} function but is DPT " + f"{ga.dpt or f'{em}.{ga.dpt_sub}'} — expected {em}.{es:03d} so Home " + "Assistant decodes it correctly.", + name=ga.name, found=ga.dpt, expected=f"{em}.{es:03d}", + )) + elif strong and ga.dpt_main != em: + findings.append(_finding( + SEVERITY_WARN, "subdpt_suspect", addr, + f"'{ga.name}' looks like a {em}.{es:03d} value but its DPT main is " + f"{ga.dpt_main} (DPT {ga.dpt or '?'}) — expected main {em}.", + name=ga.name, found=ga.dpt, expected=f"{em}.{es:03d}", + )) + return findings + + +# --------------------------------------------------------------------------- # +# KNX Secure posture + keyring handover checklist (A4). +# Report-only: this server never handles key material — it only summarises the +# per-GA Security flag and emits the ETS/HA keyring workflow as a checklist. +# --------------------------------------------------------------------------- # +def secure_posture(project: LoadedProject) -> dict[str, Any]: + """Summarise KNX Data Secure posture and the keyring handover steps.""" + gas = [g for g in project.gas.values() if g.intent == INTENT_FUNCTIONAL] + secured = [g for g in gas if g.data_secure] + plain = [g for g in gas if not g.data_secure] + + # A middle group carrying BOTH secured and plaintext GAs is a posture gap: + # a function is only as secure as its weakest address. + by_mid: dict[tuple, dict[str, int]] = defaultdict(lambda: {"sec": 0, "plain": 0}) + for g in gas: + if g.main is None: + continue + by_mid[(g.main, g.middle)]["sec" if g.data_secure else "plain"] += 1 + mixed = [{"main": k[0], "middle": k[1], "secured": v["sec"], "plaintext": v["plain"]} + for k, v in sorted(by_mid.items()) if v["sec"] and v["plain"]] + + keyring_required = len(secured) > 0 + total = len(gas) + pct = (100 * len(secured) // total) if total else 0 + + checklist = [ + "Assign every KNX Data Secure device to a **secure** tunnel/IP endpoint in " + "ETS (Project → Security).", + "Export the ETS **Keyring** (`.knxkeys`): Project → Security → Export Keyring " + "(protect it with a strong password).", + "Import the `.knxkeys` into the reader (Home Assistant KNX integration / the " + "IP interface) — Data Secure GAs cannot be read without it.", + "After **any** change to a secured GA, device or the secure topology, " + "**re-export and re-import** the keyring.", + "Store the keyring and the project password securely; **never commit them to " + "Git** or share them in plain text.", + ] + if mixed: + checklist.insert(1, f"Review the **{len(mixed)} middle group(s) with mixed " + "secure/plaintext addresses** — a function is only as " + "secure as its weakest GA; secure the whole function or none.") + + return { + "total_functional_gas": total, + "secured": len(secured), + "plaintext": len(plain), + "secured_pct": pct, + "keyring_required": keyring_required, + "mixed_middle_groups": mixed, + "secured_addresses": [{"address": g.address, "name": g.name} for g in secured[:200]], + "checklist": checklist, + } diff --git a/nickol_knx_mcp/handover.py b/nickol_knx_mcp/handover.py index 3a1b730..a590db6 100644 --- a/nickol_knx_mcp/handover.py +++ b/nickol_knx_mcp/handover.py @@ -18,7 +18,7 @@ from html import escape from typing import Any from .project import LoadedProject -from .analyze import validate_naming, detect_missing_status, detect_dpt_issues +from .analyze import validate_naming, detect_missing_status, detect_dpt_issues, secure_posture from .intent import INTENT_FUNCTIONAL @@ -219,14 +219,27 @@ def build_handover(project: LoadedProject, "state for these until a feedback GA is added.\n" ) - # 5. KNX Secure - md.append("\n## 5. KNX Secure scope\n") - if secure: + # 5. KNX Secure posture + keyring checklist + md.append("\n## 5. KNX Secure posture\n") + posture = secure_posture(project) + if posture["keyring_required"]: md.append( - f"**{len(secure)}** group address(es) carry the KNX Data Secure flag. " - "The next engineer needs the **ETS Keyring export (`.knxkeys`)** to " - "commission or read these — it is handled in ETS/HA, never by this tool.\n" + f"- Secured (KNX Data Secure): **{posture['secured']}** GA " + f"({posture['secured_pct']}%) · plaintext: **{posture['plaintext']}** GA\n" ) + if posture["mixed_middle_groups"]: + md.append( + f"- ⚠️ **{len(posture['mixed_middle_groups'])} middle group(s) mix " + "secured + plaintext addresses** — a function is only as secure as its " + "weakest GA. Secure the whole function or none:\n" + ) + for mx in posture["mixed_middle_groups"][:20]: + md.append(f" - `{mx['main']}/{mx['middle']}/*` — " + f"{mx['secured']} secured, {mx['plaintext']} plaintext") + md.append("\n**Keyring handover checklist** (key material stays in ETS/HA — " + "never in this tool):\n") + for i, step in enumerate(posture["checklist"], 1): + md.append(f"{i}. {step}") md.append("\n
Secured group addresses\n") for ga in secure[:200]: md.append(f"- `{ga.address}` {ga.name}") @@ -234,7 +247,7 @@ def build_handover(project: LoadedProject, md.append(f"- … and {len(secure)-200} more") md.append("
\n") else: - md.append("No KNX Data Secure group addresses in this project.\n") + md.append("No KNX Data Secure group addresses — no keyring required.\n") # 6. QA state at handover — itemised findings md.append("\n## 6. QA state at handover\n") diff --git a/nickol_knx_mcp/server.py b/nickol_knx_mcp/server.py index 053c9f7..b08aa14 100644 --- a/nickol_knx_mcp/server.py +++ b/nickol_knx_mcp/server.py @@ -16,7 +16,8 @@ from typing import Any, Optional from mcp.server.fastmcp import FastMCP from .project import load_project as load_project_file, LoadedProject -from .analyze import validate_naming, detect_missing_status, detect_dpt_issues +from .analyze import (validate_naming, detect_missing_status, detect_dpt_issues, + secure_posture) from .generate_ha import generate_ha_yaml from .generate_ets import generate_ets_csv, generate_ets_xml from .report import build_report @@ -163,6 +164,18 @@ def check_dpt() -> list[dict[str, Any]]: return detect_dpt_issues(_project()) +@mcp.tool() +def check_secure() -> dict[str, Any]: + """Summarise KNX Data Secure posture + the keyring handover checklist. + + Reports how many group addresses are secured vs plaintext, flags middle + groups that mix secure and plaintext addresses (a function is only as secure + as its weakest GA), and emits the ETS/HA keyring workflow as a checklist. + Report-only — this server never touches key material. + """ + return secure_posture(_project()) + + @mcp.tool() def analyze_all(name_regex: Optional[str] = None) -> dict[str, Any]: """Run every check and return the report summary plus all findings.""" diff --git a/pyproject.toml b/pyproject.toml index 9963785..13183cc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "nickol-knx-mcp" -version = "0.3.0" +version = "0.4.0" description = "Design-time KNX/ETS6 project assistant as an MCP server (parse .knxproj, validate, generate HA YAML + ETS CSV/XML). No live bus access." readme = "README.md" requires-python = ">=3.10" diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index eb1679a..756c102 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -450,7 +450,7 @@ _hmd = _hp["markdown"] # all seven sections present for _sec in ("# KNX Handover Pack", "## 1. Topology", "## 2. Equipment inventory", "## 3. Group-address map by domain", "## 4. Command / status coverage", - "## 5. KNX Secure scope", "## 6. QA state at handover", "## 7. Pack contents"): + "## 5. KNX Secure posture", "## 6. QA state at handover", "## 7. Pack contents"): assert _sec in _hmd, f"handover missing section: {_sec}" # equipment inventory picked up the synthetic MDT device + its order number assert "MDT" in _hmd and "AKK-0416.03" in _hmd, "device inventory not rendered" @@ -487,3 +487,34 @@ _miss=decompose_device("totally-unknown-xyz") assert _miss["matched"] is False assert len(list_recipes())>=10 print("OK: decompose_device — dimmer 3.007+5.001, shutter 1.008, panel=0, unknown handled") + +# --------------------------------------------------------------------------- # +# Regression (v0.4.0): A1 sub-DPT linter + A4 KNX Secure posture. +# --------------------------------------------------------------------------- # +print("\n=== REGRESSION: A1 sub-DPT linter + A4 secure posture ===") +from nickol_knx_mcp.analyze import secure_posture +_rawA1 = {"group_addresses": { + "5/0/0": ga("5/0/0", "Гостиная - Температура воздуха", 9, 4), # temp but 9.004(lux) -> suspect + "5/0/1": ga("5/0/1", "Спальня - Температура", 9, 1), # correct -> ok + "1/0/0": ga("1/0/0", "Kitchen brightness value", 5, 10), # brightness but 5.010 -> suspect + "1/0/1": ga("1/0/1", "Hall brightness value", 5, 1), # correct + "2/0/0": ga("2/0/0", "CO2 living room", 5, 1), # co2 (strong) wrong main -> suspect +}} +_sd = {(f["code"], f["address"]) for f in detect_dpt_issues(build_loaded_from_raw(_rawA1, "mem"))} +assert ("subdpt_suspect", "5/0/0") in _sd, "temp 9.004 not caught" +assert ("subdpt_suspect", "1/0/0") in _sd, "brightness 5.010 not caught" +assert ("subdpt_suspect", "2/0/0") in _sd, "co2 wrong-main not caught" +assert ("subdpt_suspect", "5/0/1") not in _sd and ("subdpt_suspect", "1/0/1") not in _sd, "false positive on correct DPT" +print("OK: sub-DPT linter — wrong sub + wrong main caught, correct DPTs clean") + +_rawS = {"group_addresses": { + "1/0/0": ga("1/0/0", "Kitchen switch", 1, 1, secure=True), + "1/0/1": ga("1/0/1", "Kitchen switch status", 1, 11, secure=False), # same middle -> mixed + "2/0/0": ga("2/0/0", "Hall switch", 1, 1, secure=True), +}} +_sp = secure_posture(build_loaded_from_raw(_rawS, "mem")) +assert _sp["secured"] == 2 and _sp["plaintext"] == 1, _sp +assert _sp["keyring_required"] is True +assert len(_sp["mixed_middle_groups"]) >= 1, "mixed secure/plaintext middle not flagged" +assert any("keyring" in s.lower() for s in _sp["checklist"]), "no keyring step" +print("OK: secure posture — counts, mixed-middle flag, keyring checklist")