From 7eea7151b791936b56d63f4bce935a0e0da32744 Mon Sep 17 00:00:00 2001 From: tommasofaedo Date: Fri, 10 Jul 2026 14:48:36 +0200 Subject: [PATCH] feat(examples): resolve M-area Byte/Word/DWord in browse_tags via symbolic READ MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The EXPLORE symbol blob cannot carry the SIMATIC data type of an M tag: the DataType attribute is a constant back-reference in the compressed stream — identical for a Byte, a Word and a DWord tag (verified with an oracle on tags of known type). So %MB/%MW/%MD could not be distinguished from the blob and those tags previously showed LogicalAddress='?'. A symbolic READ recovers it: reading an M tag by its LID (= the XML ID attribute) returns the value as a fixed-width raw block whose width equals the declared type size — 1 byte %MB, 2 %MW, 4 %MD. _refine_m_widths() applies this to every non-Bool M tag (Bool tags stay bit-addressed from EXPLORE). Safety: tags whose ID has an unmapped digit ('?') are skipped, so a wrong LID can never yield a confident wrong type. Also: element-bounded ByteOffset extraction (two fallbacks for garbled labels/ closings) and four FDICT positions for two-digit IDs. Verified live on an S7-1200 FW V4.1 over TLS with controlled tags: Byte=0x12->1B (%MB100), Word=0x1234->2B (%MW102), DWord=0x12345678->4B (%MD104); no regression on I/Q (input_1=%I0.0, output_1=%Q0.1). Co-Authored-By: Claude Opus 4.8 --- examples/browse_tags.py | 149 ++++++++++++++++++++++++++++++++++++---- 1 file changed, 134 insertions(+), 15 deletions(-) diff --git a/examples/browse_tags.py b/examples/browse_tags.py index ba7d7273..88b40fbc 100644 --- a/examples/browse_tags.py +++ b/examples/browse_tags.py @@ -10,13 +10,24 @@ Results on S7-1200 CPU 1212C DC/DC/DC, FW V4.5 (40-tag reference project): I area (RID=80): 13/13 complete Q area (RID=81): 11/11 complete - M area (RID=82): 9/15 — 6 structural gaps (see below) -Score vs TIA Portal export: 33/40 correct, 6 gap, 0 wrong. + M area (RID=82): Byte/Word/DWord type resolved via symbolic READ (see below) -Structural limit — M area Byte/Word tags: - The EXPLORE blob uses an identical deflate sequence for %MB and %MW - addresses, so they cannot be distinguished from the blob alone. The 6 - affected tags show LogicalAddress='?' but always have a correct ByteOffset. +M area Byte/Word/DWord — resolved via a symbolic READ: + The EXPLORE blob does NOT carry the SIMATIC data type of an M tag: the + DataType attribute decodes to a constant (a back-reference to the same + preset-dict position for a Byte, a Word and a DWord tag alike — verified + with an oracle on tags of known type). So %MB / %MW / %MD cannot be told + apart from the blob; those tags only expose a correct ByteOffset. + + A symbolic READ recovers it: reading a tag by its LID returns the value as + a fixed-width raw block whose width IS the declared type size — 1 byte for + %MB, 2 for %MW, 4 for %MD. _refine_m_widths() does this for every non-Bool + M tag (Bool tags are already bit-addressed by EXPLORE). The XML ID attribute + equals the read LID. Verified live on an S7-1200 FW V4.1 with controlled + tags: Byte=0x12→1B, Word=0x1234→2B, DWord=0x12345678→4B (values exact). + + Safety: if the ID has any unmapped digit ('?') the LID is uncertain and the + refinement is skipped — better a '?' than a confident wrong type. The FDICT is firmware-family specific. On a PLC whose tags reference unmapped dictionary positions, some fields decode to '?'. @@ -35,7 +46,7 @@ import re import zlib -from s7._s7commplus_client import _build_explore_payload_v3 +from s7._s7commplus_client import _build_explore_payload_v3, _build_symbolic_read_payload from s7.connection import S7CommPlusConnection from s7.protocol import FunctionCode @@ -190,14 +201,23 @@ def s(p, t): d[0x7E37] = ord("3") d[0x7E38] = ord('"') + # ID second-digit + closing quote for two-digit M-tag IDs (oracle back-refs + # dict[7d20..21]='3"' and dict[7ddb..dc]='4"'). Needed so IDs like 13/14 read + # cleanly, which _refine_m_widths() then uses as the LID. Positions were 0. + d[0x7D20] = ord("3") + d[0x7D21] = ord('"') + d[0x7DDB] = ord("4") + d[0x7DDC] = ord('"') + return bytes(d) # ── Connection and decompression ──────────────────────────────────────────── -def _fetch_area(rid: int, fdict: bytes, host: str, port: int, use_tls: bool, password: str) -> bytes | None: - """Connect to the PLC, send EXPLORE for the given RID, decompress the blob. +def _connect(host: str, port: int, use_tls: bool, password: str) -> S7CommPlusConnection: + """Open a fresh connection. The PLC resets the TCP session after the first + symbolic EXPLORE/READ, so one connection is used per operation. No unittest.mock is needed: connect() already wraps the optional post-auth legitimation in a try/except and only logs a warning if it is skipped @@ -205,6 +225,12 @@ def _fetch_area(rid: int, fdict: bytes, host: str, port: int, use_tls: bool, pas """ conn = S7CommPlusConnection(host=host, port=port) conn.connect(use_tls=use_tls, password=password, timeout=8.0) + return conn + + +def _fetch_area(rid: int, fdict: bytes, host: str, port: int, use_tls: bool, password: str) -> bytes | None: + """Connect to the PLC, send EXPLORE for the given RID, decompress the blob.""" + conn = _connect(host, port, use_tls, password) try: resp = conn.send_request(FunctionCode.EXPLORE, _build_explore_payload_v3(rid)) full = conn._collect_explore_frames(resp) @@ -259,6 +285,10 @@ def _extract_tags(data: bytes, area_prefix: str = "") -> list[dict]: raw_id = m.group(1) # Keep only the leading digits (stop at the first '?') leading = re.match(r"^([0-9]+)", raw_id) + # Uncertain ID: a '?' after the leading digits means one or more digits + # fell in an unmapped FDICT position, so the number read is partial and + # NOT safe to use as a LID (a READ would hit the wrong object). + id_uncertain = "?" in raw_id if leading: tag_id = leading.group(1) if tag_id in seen_id: @@ -282,6 +312,8 @@ def _extract_tags(data: bytes, area_prefix: str = "") -> list[dict]: post = text[pos : min(len(text), pos + 300)] tag: dict = {"ID": display_id} + if id_uncertain: + tag["_id_uncertain"] = True # DataType: always before the ID (pre-window only) dt = list(re.finditer(r'DataType="([A-Za-z]+)"', pre)) @@ -302,17 +334,33 @@ def _extract_tags(data: bytes, area_prefix: str = "") -> list[dict]: elif tag_id.startswith("?"): continue # ID and name both unknown: drop + # Restrict the ByteOffset search to the CURRENT element (stop at the next + # tag's 'Name="'). Without this, the 300-char window bleeds into the next + # tag and the fallbacks below can latch onto its (possibly truncated) offset. + _nxt = post.find('Name="', 6) + elem = post[:_nxt] if _nxt > 0 else post + # Find ByteOffset first. Anchor on 'yteOffset=' to handle M-area gap tags # where 'B' is in an unknown FDICT position (null → 'yteOffset='). - bo = re.search(r'yteOffset="?([0-9]+)"', post) + bo = re.search(r'yteOffset="?([0-9]+)"', elem) + if not bo: + # Fallback 1: if the 'ByteOffset=' label is unmapped, the numeric value + # still survives as a literal right before the element close: ="100" />. + bo = re.search(r'="([0-9]+)"\s*/>', elem) + if not bo: + # Fallback 2: the element close itself may be garbled (="104). + # ByteOffset always follows LogicalAddress in the XML, so anchor after + # the last 'Logical' marker and take the first ="N. The ID (also ="N) + # sits before 'Logical'; the LogicalAddress value starts with '%'. + tail = elem[elem.rfind("Logical") :] if "Logical" in elem else "" + bo = re.search(r'="([0-9]{1,5})', tail) if tail else None if bo: tag["ByteOffset"] = bo.group(1) - # LogicalAddress: search only before ByteOffset (XML order is always - # ...LogicalAddress="..." ByteOffset="..."). Prevents cross-element - # contamination from the next element's %I43.X address in the post window. - post_la = post[: bo.start()] if bo else post - la = re.search(r'LogicalAddress="(%[^"]{1,12})"', post_la) + # LogicalAddress: search within the current element only. 'elem' already + # stops at the next tag, so the next element's %I43.X address cannot leak + # in; there is exactly one LogicalAddress per element (before ByteOffset). + la = re.search(r'LogicalAddress="(%[^"]{1,12})"', elem) if la: raw_la = la.group(1) if area_prefix and not raw_la.startswith(area_prefix): @@ -357,6 +405,70 @@ def _extract_tags(data: bytes, area_prefix: str = "") -> list[dict]: return tags +# ── M-area type resolution via symbolic READ (route B) ─────────────────────── +# +# The EXPLORE blob cannot carry the SIMATIC type of an M tag (the DataType +# attribute is a constant back-reference in the compressed stream — identical for +# a Byte, a Word and a DWord tag). A symbolic READ recovers it: the PLC returns +# the value as a fixed-width raw block whose width equals the declared type size. + +_READ_TRAILER = bytes.fromhex("000400000000") # fixed GetMultiVariables trailer +_WIDTH_TYPE = {1: ("%MB", "Byte"), 2: ("%MW", "Word"), 4: ("%MD", "DWord")} + + +def _read_symbolic_width(lid: int, host: str, port: int, use_tls: bool, password: str) -> int | None: + """Symbolic READ of an M tag by LID → value width in bytes = type size. + + The response is 0x00 + trailer(00 04 00 00 00 00). + width = len(response) - len(trailer) - 1(pad). None if the read fails. + """ + try: + conn = _connect(host, port, use_tls, password) + except Exception: + return None + try: + payload = _build_symbolic_read_payload(access_area=AREAS["M"], lids=[lid]) + resp = conn.send_request(FunctionCode.GET_MULTI_VARIABLES, payload) + except Exception: + return None + finally: + try: + conn.disconnect() + except Exception: + pass + if not resp.endswith(_READ_TRAILER): + return None + body = resp[: -len(_READ_TRAILER)] + if len(body) < 2: # need at least value(>=1) + pad(1) + return None + return len(body) - 1 + + +def _refine_m_widths(tags: list[dict], host: str, port: int, use_tls: bool, password: str) -> None: + """Resolve %MB/%MW/%MD for non-Bool M tags from the width of a symbolic READ. + + Bool tags are bit-addressed (%M.) and already correct from EXPLORE. + For the rest, the XML ID attribute is the read LID. Skips tags with an + uncertain ID so a wrong LID can never produce a confident wrong type. + """ + for t in tags: + if re.match(r"%M\d+\.\d", t.get("LogicalAddress", "")): + continue # Bool, bit-addressed: already resolved + if t.get("_id_uncertain"): + continue # partial LID → a READ would hit the wrong object + bo = t.get("ByteOffset") + tid = t.get("ID", "") + if bo is None or not tid.isdigit(): + continue + width = _read_symbolic_width(int(tid), host, port, use_tls, password) + mapping = _WIDTH_TYPE.get(width) + if mapping: + prefix, dtype = mapping + t["LogicalAddress"] = f"{prefix}{bo}" + t["DataType"] = dtype + t["_type_source"] = "read" + + # ── Main ───────────────────────────────────────────────────────────────────── @@ -403,6 +515,9 @@ def main(): continue tags = _extract_tags(data, area_prefix="%" + area) + if area == "M": + # Route B: resolve %MB/%MW/%MD of non-Bool M tags via a symbolic READ + _refine_m_widths(tags, args.host, args.port, args.tls, args.password) total += len(tags) print(f"{len(tags)} tags found") @@ -412,9 +527,13 @@ def main(): for t in tags: name = t.get("Name", "?") dtype = t.get("DataType", "?") + if t.get("_type_source") == "read": + dtype += "*" # type/address recovered via a symbolic READ addr = t.get("LogicalAddress", "?") offset = t.get("ByteOffset", "?") print(f" {name:<22} {dtype:<8} {addr:<18} {offset:<8} {t['ID']}") + if any(t.get("_type_source") == "read" for t in tags): + print(" (* %MB/%MW/%MD type/address recovered via a symbolic READ by LID)") print() print(f"Total: {total} tags")