diff --git a/examples/browse_tags.py b/examples/browse_tags.py
index 56f97dc0..ba7d7273 100644
--- a/examples/browse_tags.py
+++ b/examples/browse_tags.py
@@ -1,54 +1,48 @@
-"""
-browse_tags.py — Read symbolic I/Q/M tags from an S7-1200 FW V4.5 PLC.
+"""browse_tags.py — Read symbolic I/Q/M tags from an S7-1200 PLC (V3 protocol).
+
+PROOF OF CONCEPT — not a supported library feature.
-Technique: EXPLORE + decompression with a partial preset dictionary
-(Adler-32 0xce9b821b, 594 of 32768 bytes reconstructed via oracle analysis).
-The same FDICT is used for all three areas (I, Q, M) — confirmed on
-independent Wireshark pcapng captures.
+Technique: EXPLORE + zlib decompression with a partial preset dictionary
+(FDICT, Adler-32 0xce9b821b) reconstructed via "oracle" analysis. The same
+FDICT is used for all three areas (I, Q, M) and is shared by S7-1200 FW V4.5
+("G1") and FW V4.1 ("G2"), confirmed on independent Wireshark pcapng captures.
-Results on S7-1200 CPU 1212C DC/DC/DC, FW V4.5 (40-tag project):
+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.
-Prerequisites:
- - python-snap7 S7CommPlus branch with Patches 1, 5, 6 applied
- (SequenceNumber field, _collect_explore_frames, session key)
- - No password, no TLS (adjust connect() call if needed)
-
-Usage:
- python browse_tags.py # all areas
- python browse_tags.py I # I area only
- python browse_tags.py Q M # Q and M
-
Structural limit — M area Byte/Word tags:
The EXPLORE blob uses an identical deflate sequence for %MB and %MW
- addresses. It is not possible to distinguish them without external
- information (e.g. a TIA Portal export). The 6 affected tags show
- LogicalAddress='?' but always have correct ByteOffset values.
+ addresses, so they cannot be distinguished from the blob alone. The 6
+ affected tags show LogicalAddress='?' but always have a correct ByteOffset.
+
+The FDICT is firmware-family specific. On a PLC whose tags reference
+unmapped dictionary positions, some fields decode to '?'.
+
+Usage:
+ python browse_tags.py 192.168.5.11 # all areas, plaintext
+ python browse_tags.py 192.168.5.11 --tls # G2 (TLS required)
+ python browse_tags.py 192.168.5.11 I Q # only I and Q areas
+ python browse_tags.py 10.0.0.5 --tls --password s3cret M
-Tested on:
- Siemens S7-1200 CPU 1212C DC/DC/DC, firmware V4.5, IP 192.168.5.11
+Requires the python-snap7 S7CommPlus support (_build_explore_payload_v3,
+collect_explore_frames) to be available in the installed `s7` package.
"""
+import argparse
import re
-import sys
import zlib
-from unittest import mock
-PLC_HOST = "192.168.5.11"
-PLC_PORT = 102
+from s7._s7commplus_client import _build_explore_payload_v3
+from s7.connection import S7CommPlusConnection
+from s7.protocol import FunctionCode
AREAS = {"I": 80, "Q": 81, "M": 82}
-# ---------------------------------------------------------------------------
-# Partial preset dictionary (594 of 32768 positions mapped, Adler-32 0xce9b821b)
-# Reconstructed via oracle technique: inflate the same blob 4 times with
-# DICT_ZERO, DICT_FF, DICT_A (i%256), DICT_B (i>>8). Identical bytes are
-# literals; differing bytes reveal FDICT position (B_out<<8)|A_out.
-# ---------------------------------------------------------------------------
+# ── Partial preset dictionary (647 of 32768 positions mapped) ────────────────
def _build_fdict() -> bytes:
@@ -58,99 +52,159 @@ def s(p, t):
for i, c in enumerate(t.encode("latin-1")):
d[p + i] = c
- # Root / element structure
- s(0x7FF1, "ControllerTags>")
+ # Leading block of a Word element (Tag_04, %IW43, ID=9) — I area
+ s(0x7DB3, 'Tags Name="Tag_04" DataType="Word" ')
+ # First Q-area element (ID=9): oracle Q pos 55-89 RUN[35] dict[7c3f..7c61]
+ # Evidence: literal '2' at pos 183 = bit offset → %Q0.2 = Tag_2
+ # Byte count: 'Tags Name="Tag_02" DataType="Bool" ' = 35c ✓
+ s(0x7C3F, 'Tags Name="Tag_02" DataType="Bool" ')
+ # ID attribute
+ s(0x7B9C, 'ID="')
+ s(0x7BA0, '9" HmiVisible="')
+ # Boolean values
+ s(0x7FC5, "True")
+ s(0x7FE7, 'Visible="')
+ s(0x7F36, 'True"')
+ # Common attribute block
+ s(0x7F71, '" HmiAccessible="True" Retain="False" Logical')
+ s(0x7FA0, 'Address="%IW"')
+ d[0x7FAC] = ord('"')
+ s(0x7FAD, ' ByteOffset="')
+ # Element separator: closes the current element, opens the next
s(0x7CC3, '" />')
d[0x7CC7] = ord("<")
s(0x7CC8, "Tags")
d[0x7CCC] = ord(" ")
- s(0x7CCD, 'HmiVisible="')
- s(0x7C0D, "/>")
+ # 88-char block for a Bool tag (split around the 0x7ed6-0x7ed7 gap)
b88 = b'" Comment="" HmiVisible="True" HmiAccessible="True" Retain="False" LogicalAddress="%I43.'
for i in range(63):
d[0x7E97 + i] = b88[i]
for i in range(25):
d[0x7ED8 + i] = b88[63 + i]
- s(0x7EF2, '" ByteOffset="') # 14c — unlocks ByteOffset for M/Q Bool
- d[0x7EF1] = ord("4") # Clock_1.25Hz bit digit → %M0.4
-
- # Common attribute blocks
- s(0x7BAF, 'True" HmiAccessible="True" Retain="False" Logical')
- s(0x7B8A, '" DataType="Bool" ID="')
- s(0x7B9C, 'ID="')
- s(0x7BA0, '9" HmiVisible="')
- s(0x7FE7, 'Visible="')
- s(0x7F36, 'True"')
- s(0x7F71, '" HmiAccessible="True" Retain="False" Logical')
- s(0x7FAD, ' ByteOffset="')
- d[0x7FAC] = ord('"')
- s(0x7FC5, "True")
+ # Close LogicalAddress + open ByteOffset attr (14c)
+ # Oracle M: after literal bit-digit, RUN[14] dict[7ef2..7eff] = '" ByteOffset="'
+ s(0x7EF2, '" ByteOffset="') # 7ef2..7eff
+
+ # Q/M Name opener: Visible="True" Name=" prefix + Tag_1 prefix
+ # Confirmed by oracle M: RUN[12] dict[7b79..7b84] precedes Clock_X names
+ s(0x7B79, 'True" Name="Tag_1') # 7b79..7b89
+ # Bool DataType+ID block for M-area tags
+ # Confirmed by oracle M: RUN[22] dict[7b8a..7b9f] after Clock_X names
+ s(0x7B8A, '" DataType="Bool" ID="') # 7b8a..7b9f
+ # "output" suffix in Q-area names (0_output, 100_output, output_0_0)
+ # Confirmed by oracle Q: RUN[7] dict[7f17..7f1d] after literal 'o'
+ s(0x7F17, 'utput" ') # 7f17..7f1d
+ # G2 FW V4.1: 'nput' for I-area tag 'input_1' (i=literal, nput=FDICT[7e50..7e53])
+ # Confirmed by oracle G2 I area: RUN[4] dict[7e50..7e53] after literal 'i'
+ s(0x7E50, "nput") # 7e50..7e53
+ # G2 FW V4.1: Q-area LogicalAddress block (25c) = same as dict[7ed8..7ef0]
+ # Confirmed by oracle G2 Q area: RUN[25] dict[7be2..7bfa] → garbled '%I43.1' → %Q0.1
+ s(0x7BE2, 'se" LogicalAddress="%I43.') # 7be2..7bfa
+ # "Tag_" prefix for M-area names (Tag_11, Tag_5, Tag_20, ...)
+ # Confirmed by oracle M: dict[7e7a..7e7d] precedes the name digit
+ s(0x7E7A, "Tag_") # 7e7a..7e7d
+ # Digit '5' + name close for tags whose second digit is 5 (Tag_5/15/25)
+ # DataType not set here: same block serves Bool and Word
+ d[0x7E7E] = ord("5") # name digit
+ d[0x7E7F] = ord('"') # closes Name="
+ # Digit '3' + name close + DataType for tags with digit 3 (Tag_3, Tag_13)
+ # Oracle Q: dict[7d08..7d1a] (19c) used twice (Tag_13 ID=13, Tag_3 ID=25)
+ d[0x7D08] = ord("3") # name digit
+ d[0x7D09] = ord('"') # closes Name="
+ s(0x7D0A, ' DataType="Bool" ') # 7d0a..7d1a
+
+ # M-area Tag_1 megablock: chars 54..102 (dict[7baf..7bdf]) = 49c
+ # = 'True" HmiAccessible="True" Retain="False" Logical'
+ # Also used by Q-area tags via dict[7ba1..7bdf] (63c) for the HmiVisible suffix
+ s(0x7BAF, 'True" HmiAccessible="True" Retain="False" Logical') # 7baf..7bdf
+
+ # First digit of the 2-digit IDs of the M-area Clock tags (ID=1X)
+ # Clock_2.5Hz→"13", Clock_2Hz→"14", Clock_1Hz→"16", Clock_0.625Hz→"17", ...
+ d[0x7C66] = ord("1")
+ # Second digit of Clock_1.25Hz ID: oracle M pos 893 → dict[7e96..7ed5] RUN[64]
+ # The '5' is NOT literal (unlike '6' for Clock_1Hz) → comes from the FDICT
+ d[0x7E96] = ord("5")
- # LogicalAddress blocks
- s(0x7FA0, 'Address="%IW"') # Word/Int I area
- s(0x7D62, 'lAddress="%QW" ByteOffset=2') # Word Q area
- s(0x7BFC, '" ByteOffset="') # Q/M Bool ByteOffset opener
- d[0x7BFB] = ord("1") # bit '1' (Clock_5Hz, Tag_25)
+ # Middle of the Q/M element separator: HmiVisible=" (12c)
+ # Q oracle: RUN[22] dict[7cc3..7cd8] = '" /> bytes | None:
- """Connect to PLC, send EXPLORE for the given RID, decompress the blob."""
- try:
- from s7._s7commplus_client import _build_explore_payload_v3
- from s7.connection import S7CommPlusConnection
- from s7.protocol import FunctionCode
- except ImportError:
- print("Error: python-snap7 not found. pip install python-snap7")
- sys.exit(1)
-
- with mock.patch.object(S7CommPlusConnection, "_post_auth_legitimation", return_value=None):
- conn = S7CommPlusConnection(host=PLC_HOST, port=PLC_PORT)
- conn.connect(use_tls=False, password="", timeout=5.0)
+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.
+
+ 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
+ (e.g. no-password PLCs, or the plaintext V3 path where legitimation needs TLS).
+ """
+ conn = S7CommPlusConnection(host=host, port=port)
+ conn.connect(use_tls=use_tls, password=password, timeout=8.0)
try:
resp = conn.send_request(FunctionCode.EXPLORE, _build_explore_payload_v3(rid))
full = conn._collect_explore_frames(resp)
@@ -160,21 +214,26 @@ def _fetch_area(rid: int, fdict: bytes) -> bytes | None:
except Exception:
pass
+ # Find the preset-dictionary blob (magic 78 7D)
p = full.find(b"\x78\x7d")
if p < 0:
return None
+ raw_deflate = full[p + 6 :] # skip 2 magic bytes + 4 dict Adler-32 bytes
try:
- return zlib.decompressobj(wbits=-15, zdict=fdict).decompress(full[p + 6 :])
+ return zlib.decompressobj(wbits=-15, zdict=fdict).decompress(raw_deflate)
except zlib.error:
return None
-# ---------------------------------------------------------------------------
-# Tag extraction
-# ---------------------------------------------------------------------------
+# ── Tag extraction ──────────────────────────────────────────────────────────
def _normalize_name(raw: str) -> str:
+ """Normalize internal PLC tag names to the TIA Portal form.
+
+ S7-1200 FW V4.5 stores names internally as 'Tag_04', 'Tag_06' (leading
+ zero), while TIA Portal exports 'Tag_4', 'Tag_6'. Strip the leading zero.
+ """
m = re.match(r"^(Tag_)0+([0-9]+)$", raw)
if m:
return m.group(1) + m.group(2)
@@ -184,14 +243,9 @@ def _normalize_name(raw: str) -> str:
def _extract_tags(data: bytes, area_prefix: str = "") -> list[dict]:
"""Extract tags from the decompressed XML blob.
- Unknown FDICT positions produce null bytes, shown as '?' after decoding.
- Extraction anchors on always-literal ID values, which are never stored
- in the preset dictionary and are therefore always visible in the output.
-
- Byte-type fallback for I/Q areas:
- Bool -> garbled %I43.X in blob -> reconstruct %{A}{bo}.{bit}
- Word -> %IW/%QW visible in blob -> append ByteOffset
- Byte -> only remaining type -> reconstruct %{A}B{ByteOffset}
+ ID values are always literal bytes in the deflate stream (always visible).
+ Anchor on each ID="N" and look for Name/DataType in the window before it.
+ Null bytes (unknown dict positions) become '?' in the text.
"""
text = data.replace(b"\x00", b"?").decode("latin-1")
tags: list[dict] = []
@@ -199,30 +253,43 @@ def _extract_tags(data: bytes, area_prefix: str = "") -> list[dict]:
seen_name: set[str] = set()
_synthetic = 0
+ # Some IDs sit in unknown dict positions (null→'?'). Capture up to 6 chars
+ # after ID=" including a possible '?' terminator.
for m in re.finditer(r'ID="([0-9?]{1,6})[?"]', text):
raw_id = m.group(1)
+ # Keep only the leading digits (stop at the first '?')
leading = re.match(r"^([0-9]+)", raw_id)
if leading:
tag_id = leading.group(1)
if tag_id in seen_id:
continue
seen_id.add(tag_id)
+ display_id = tag_id
else:
+ # ID entirely in FDICT (all '?') → synthetic key
tag_id = f"?{_synthetic}"
_synthetic += 1
+ display_id = "?"
pos = m.start()
+
+ # Use the last list[dict]:
seen_name.add(normalized)
tag["Name"] = normalized
elif tag_id.startswith("?"):
- continue
+ continue # ID and name both unknown: drop
+ # 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)
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)
if la:
raw_la = la.group(1)
if area_prefix and not raw_la.startswith(area_prefix):
- garbled = re.match(r"%I43\.([0-7])", raw_la)
- if garbled:
- tag["_garbled_bit"] = garbled.group(1)
+ if area_prefix in ("%Q", "%M"):
+ garbled = re.match(r"%I43\.([0-7])", raw_la)
+ if garbled:
+ tag["_garbled_bit"] = garbled.group(1)
elif not re.search(r"\?{3,}", raw_la):
- tag["LogicalAddress"] = raw_la.rstrip("?")
+ # I area: the '43' in '%I43.x' comes from the FDICT sample dict,
+ # NOT the real tag (same source as the %Q/%M garbled path). Only the
+ # bit digit is a real literal. If ByteOffset disagrees, rebuild
+ # %I{ByteOffset}.{bit}. Fixes G2 input_1: %I43.0 → %I0.0.
+ mi = re.match(r"%I43\.([0-7])$", raw_la)
+ if area_prefix == "%I" and mi and tag.get("ByteOffset") not in (None, "43"):
+ tag["_garbled_bit"] = mi.group(1)
+ else:
+ tag["LogicalAddress"] = raw_la.rstrip("?")
if "_garbled_bit" in tag:
bit = tag.pop("_garbled_bit")
if "ByteOffset" in tag:
+ # The bit digit in the garbled %I43.X address is always literal and
+ # correct — always use the bit form %{area}{bo}.{bit}.
area_letter = area_prefix[1]
tag["LogicalAddress"] = f"%{area_letter}{tag['ByteOffset']}.{bit}"
+ # Byte-type fallback for I/Q: Bool tags trigger _garbled_bit above,
+ # Word/Int tags have %IW/%QW visible. The only remaining case is Byte
+ # (%IB / %QB): oracle analysis proved the value is not encoded. Rebuild it.
if "LogicalAddress" not in tag and "ByteOffset" in tag:
if area_prefix in ("%I", "%Q"):
area_letter = area_prefix[1]
tag["LogicalAddress"] = f"%{area_letter}B{tag['ByteOffset']}"
+ # Reconstruct %IW/%QW/%MW → %IW{N} etc. using ByteOffset.
la_val = tag.get("LogicalAddress", "")
if la_val and "ByteOffset" in tag and re.match(r"^%[A-Z]{2,}$", la_val):
tag["LogicalAddress"] = f"{la_val}{tag['ByteOffset']}"
@@ -270,29 +357,49 @@ def _extract_tags(data: bytes, area_prefix: str = "") -> list[dict]:
return tags
-# ---------------------------------------------------------------------------
-# Main
-# ---------------------------------------------------------------------------
+# ── Main ─────────────────────────────────────────────────────────────────────
def main():
- areas_requested = [a.upper() for a in sys.argv[1:] if a.upper() in AREAS]
- if not areas_requested:
- areas_requested = list(AREAS)
+ parser = argparse.ArgumentParser(
+ description="Read symbolic I/Q/M tags from an S7-1200 PLC (V3 protocol).",
+ )
+ parser.add_argument("host", help="PLC IP address or hostname")
+ parser.add_argument(
+ "areas",
+ nargs="*",
+ default=None,
+ help="Areas to browse: I, Q and/or M (default: all)",
+ )
+ parser.add_argument("--port", type=int, default=102, help="TCP port (default: 102)")
+ parser.add_argument(
+ "--tls",
+ action="store_true",
+ help="Activate TLS (required on PLCs that enforce secure PG/PC comms, e.g. G2 FW V4.1)",
+ )
+ parser.add_argument("--password", default="", help="PLC password (default: none)")
+ args = parser.parse_args()
+
+ requested = [a.upper() for a in args.areas if a.upper() in AREAS] if args.areas else list(AREAS)
+ if not requested:
+ requested = list(AREAS)
fdict = _build_fdict()
- non_zero = sum(1 for b in fdict if b)
- print(f"PLC: {PLC_HOST}:{PLC_PORT}")
- print(f"FDICT: {non_zero} positions mapped of 32768 (Adler-32: 0xce9b821b)\n")
+ print(f"PLC: {args.host}:{args.port} (tls={args.tls})")
+ print(f"Partial FDICT: {sum(1 for b in fdict if b)} of 32768 positions known\n")
total = 0
- for area in areas_requested:
+ for area in requested:
rid = AREAS[area]
print(f">> Area {area} (RID={rid}) ... ", end="", flush=True)
- data = _fetch_area(rid, fdict)
+ try:
+ data = _fetch_area(rid, fdict, args.host, args.port, args.tls, args.password)
+ except Exception as exc:
+ print(f"CONNECTION ERROR: {exc!r}")
+ continue
if data is None:
- print("ERROR: compressed blob not found or decompression failed")
+ print("ERROR: no compressed blob found")
continue
tags = _extract_tags(data, area_prefix="%" + area)
@@ -307,14 +414,10 @@ def main():
dtype = t.get("DataType", "?")
addr = t.get("LogicalAddress", "?")
offset = t.get("ByteOffset", "?")
- tid = t["ID"]
- print(f" {name:<22} {dtype:<8} {addr:<18} {offset:<8} {tid}")
+ print(f" {name:<22} {dtype:<8} {addr:<18} {offset:<8} {t['ID']}")
print()
print(f"Total: {total} tags")
- print()
- print('Note: 6 M-area tags (Byte/Word type) show LogicalAddress="?" —')
- print("structural limit of the S7CommPlus EXPLORE protocol (see module docstring).")
if __name__ == "__main__":