mirror of
https://github.com/NickoScope/nickol-knx-mcp.git
synced 2026-09-29 19:31:12 +02:00
feat: sub-DPT linter + KNX Secure posture (v0.4.0)
A1 — sub-DPT sanity linter: when a GA name implies a specific DPT sub-type (temperature->9.001, power->14.056, brightness/position->5.001), flag a wrong sub or wrong main. Multilingual, conservative; surfaced as the subdpt_suspect finding via check_dpt / analyze_all. A4 — KNX Data Secure posture: secure_posture() + new check_secure MCP tool. Report-only summary (secured vs plaintext counts, mixed secure/plaintext middle groups, keyring handover checklist). Reads only the per-GA Security flag; no key material touched. Handover pack section 5 rewritten to this posture section. Version bump to 0.4.0; CHANGELOG updated. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
b16492eb68
commit
44e1ae3aa7
@@ -1,2 +1,2 @@
|
||||
"""nickol-knx-mcp: design-time KNX/ETS project assistant MCP server."""
|
||||
__version__ = "0.3.0"
|
||||
__version__ = "0.4.0"
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
@@ -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<details><summary>Secured group addresses</summary>\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("</details>\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")
|
||||
|
||||
@@ -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."""
|
||||
|
||||
Reference in New Issue
Block a user