localeye.cam

← Back

tools/wx_probe.py

doors version 2.5 · 12 KB

1#!/usr/bin/env python3
2"""
3wx_probe.py - standalone ICT Protege WX connection tester (does NOT touch the NVR).
4
5Run it on any machine that can reach the controller:
6
7 pip install requests cryptography
8 python3 wx_probe.py http://192.168.40.20:82 juspitt 'your-password'
9
10It walks the login handshake one step at a time and prints the raw reply from the controller at each step,
11plainly labelled, so we can see exactly where it goes wrong. It tries the request both PLAIN and ENCRYPTED
12(the NVR uses encrypted), and reports which the controller accepts.
13
14Nothing is written anywhere; it only reads. Add -v to also print the raw hex sent.
15"""
16import argparse
17import hashlib
18import random
19import sys
20
21try:
22 import requests
23except ImportError:
24 sys.exit("Need 'requests': pip install requests")
25try:
26 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
27except ImportError:
28 sys.exit("Need 'cryptography': pip install cryptography")
29
30DLL = "PRT_CTRL_DIN_ISAPI.dll"
31DEFAULT_KEY = "12345678911234567892123456789312" # WX built-in handshake key (full 32 chars)
32
33
34# ------------------------------------------------------------------ crypto (mirrors the controller's JS)
35def _key_bytes(key):
36 k = key.encode("latin-1")
37 for size in (16, 24, 32):
38 if len(k) <= size:
39 return k + b"\0" * (size - len(k))
40 raise ValueError("key too long")
41
42
43def enc(text, key, sessid="", iv=None):
44 iv = iv or "".join(chr(int(75 * random.random() + 48)) for _ in range(16))
45 ib = iv.encode("latin-1")
46 n = 16 - len(text.encode("utf-8")) % 16
47 padded = text.encode("utf-8") + bytes([n]) * n
48 e = Cipher(algorithms.AES(_key_bytes(key)), modes.CBC(ib)).encryptor()
49 return sessid + (ib + e.update(padded) + e.finalize()).hex().upper()
50
51
52def dec(hextext, key):
53 raw = bytes.fromhex(hextext.strip())
54 iv, body = raw[:16], raw[16:]
55 body += b"\0" * (-len(body) % 16)
56 d = Cipher(algorithms.AES(_key_bytes(key)), modes.CBC(iv)).decryptor()
57 out = (d.update(body) + d.finalize()).rstrip(b"\0")
58 if out:
59 out = out[:-out[-1]] if 0 < out[-1] <= 16 else out
60 return out.decode("utf-8", "replace")
61
62
63# ------------------------------------------------------------------ login maths
64def wx_xor(text, number):
65 bits = bin(int(number))[2:].rjust(32, "0")
66 pos, out = len(bits), []
67 for ch in text.encode("utf-8"):
68 pos = len(bits) - 8 if pos == 0 else pos - 8
69 chunk = bits[pos:pos + 8] if pos >= 0 else bits[max(len(bits) + pos, 0):][:8]
70 out.append(f"{ch ^ (int(chunk, 2) if chunk else 0):02x}")
71 return "".join(out).upper()
72
73
74def sha1_js(text):
75 units = text.encode("utf-16-le")
76 return hashlib.sha1(bytes(units[0::2])).hexdigest()
77
78
79# ------------------------------------------------------------------ helpers
80def show(label, text):
81 body = text if len(text) <= 300 else text[:300] + f"… ({len(text)} chars)"
82 print(f" {label}: {body!r}")
83
84
85def looks_numeric(s):
86 return s.strip().lstrip("-").isdigit()
87
88
89def main():
90 ap = argparse.ArgumentParser()
91 ap.add_argument("host", help="e.g. http://192.168.40.20:82")
92 ap.add_argument("username")
93 ap.add_argument("password", nargs="?", help="omit to be prompted securely")
94 ap.add_argument("-v", "--verbose", action="store_true", help="print raw hex sent")
95 ap.add_argument("--no-verify", action="store_true", help="skip TLS verify (for https self-signed)")
96 args = ap.parse_args()
97
98 if not args.password:
99 import getpass
100 args.password = getpass.getpass("Protege password: ")
101 base = args.host.rstrip("/")
102 if not base.startswith("http"):
103 base = "http://" + base
104 url = f"{base}/{DLL}"
105 s = requests.Session()
106 s.verify = not args.no_verify
107 sessid = "".join(random.choice("0123456789ABCDEF") for _ in range(32))
108 s.cookies.set("SESSID", sessid)
109 print(f"\n=== Protege WX probe ===\nURL: {url}\nSESSID: {sessid}\nuser: {args.username}\n")
110
111 def post(payload):
112 if args.verbose:
113 print(f" -> POST body: {payload[:80]}{'…' if len(payload) > 80 else ''}")
114 r = s.post(url, data=payload, headers={"Content-Type": "text/plain; charset=utf-8"}, timeout=10)
115 r.raise_for_status()
116 return r.text
117
118 # --- Step 0: can we reach it at all?
119 print("Step 0: reachability")
120 try:
121 r = s.get(base + "/", timeout=8)
122 print(f" GET / -> HTTP {r.status_code}")
123 except Exception as e:
124 print(f" GET / FAILED: {e}\n -> wrong address/port, or a firewall is blocking it.")
125 return
126
127 # --- Step 1: InitSession, PLAIN (this is what your curl did)
128 init_cmd = f"Command&Type=Session&SubType=InitSession&SessionID={sessid}"
129 print("\nStep 1: InitSession (PLAIN text)")
130 challenge_plain = None
131 try:
132 out = post(init_cmd)
133 show("reply", out)
134 if looks_numeric(out):
135 challenge_plain = out.strip()
136 print(" -> controller accepts PLAIN requests")
137 except Exception as e:
138 print(f" FAILED: {e}")
139
140 # --- Step 2: InitSession, ENCRYPTED (this is what the NVR does)
141 print("\nStep 2: InitSession (ENCRYPTED with the default key)")
142 challenge_enc = None
143 try:
144 out = post(enc(init_cmd, DEFAULT_KEY, sessid))
145 show("reply", out)
146 if looks_numeric(out):
147 challenge_enc = out.strip()
148 print(" -> controller accepts ENCRYPTED requests (this is what the NVR needs)")
149 elif out.startswith("<invalid session>"):
150 print(" -> controller wants a different handshake (invalid-session).")
151 except Exception as e:
152 print(f" FAILED: {e}")
153
154 challenge = challenge_enc or challenge_plain
155 use_plain = challenge_enc is None and challenge_plain is not None
156 if challenge is None:
157 print("\nRESULT: the controller never returned a challenge number. Nothing else can run until it does.")
158 print("Copy everything above and send it back.")
159 return
160 if use_plain:
161 print("\nNOTE: this controller only answered PLAIN, not encrypted. That's important - tell me this.")
162
163 # --- Step 3: CheckPassword
164 print(f"\nStep 3: CheckPassword (challenge = {challenge})")
165 name = sha1_js(wx_xor(args.username.lower(), int(challenge) + 1)).upper()
166 pw = sha1_js(wx_xor(sha1_js(args.password), int(challenge))).upper()
167 check = f"Command&Type=Session&SubType=CheckPassword&Name={name}&Password={pw}&SessionID={sessid}"
168 try:
169 out = post(check if use_plain else enc(check, DEFAULT_KEY, sessid))
170 show("reply", out)
171 if out.startswith("FAIL"):
172 if "No valid operator" in out:
173 print(" -> username or password is wrong (controller reached fine).")
174 else:
175 print(" -> CheckPassword refused.")
176 return
177 if not looks_numeric(out):
178 print(" -> unexpected reply; send this back.")
179 return
180 reply = out.strip()
181 print(" -> password accepted")
182 except Exception as e:
183 print(f" FAILED: {e}")
184 return
185
186 # --- Step 4: derive the session key, then confirm the key works and read events
187 key = sha1_js(wx_xor(sha1_js(args.password), int(reply))).upper()[:16]
188 print(f"\nStep 4: session key derived ({key})")
189 from datetime import datetime, timedelta
190
191 def enc_req(cmd):
192 # post-login requests: append &Sequence= (empty, as WX does) then encrypt with the derived key
193 return enc(cmd + "&Sequence=", key, sessid)
194
195 # 4a: a simple encrypted command first, to prove the key + session work at all
196 print("\nStep 4a: GetOperatorDetails (encrypted, POST) - proves the session key works")
197 try:
198 out = post(enc_req("Command&Type=Session&SubType=GetOperatorDetails"))
199 if out.startswith("<invalid session>"):
200 show("reply", out)
201 print(" -> session/key rejected. This is the key thing to solve; details below.")
202 else:
203 dec_out = dec(out, key) if not out.startswith("<") else out
204 show("decrypted reply", dec_out[:120])
205 print(" -> session key WORKS.")
206 except Exception as e:
207 print(f" FAILED: {e}")
208
209 # --- Step 5: LIVE event feed (the real protocol, from monitorevent.js)
210 # On a controller the "All Events" page does NOT use ExportCSV. It calls:
211 # Request&Type=Events&SubType=Latest (first load)
212 # Request&Type=Events&SubType=Update (poll, new events since LastId)
213 # Request&Type=Events&SubType=Previous (older events)
214 # The reply is the WX key=value format (decrypted), starting with "Event", where each event is
215 # Event<n> = <23-char timestamp><description sentence>, plus TypeId<n>, User<n>, LastId, etc.
216 print("\n" + "=" * 60)
217 print("Step 5: LIVE event feed (Request&Type=Events&SubType=Latest)")
218 print("=" * 60)
219
220 def enc_get(cmd):
221 r = s.get(url, params=enc(cmd + "&Sequence=", key, sessid), timeout=20)
222 r.raise_for_status()
223 t = r.text
224 if t.startswith("<invalid"):
225 return t
226 try:
227 return dec(t, key)
228 except Exception:
229 return t
230
231 def parse_kv(text):
232 """WX key=value&key=value -> dict (same shape monitorevent.js consumes)."""
233 out = {}
234 for pair in text.split("&"):
235 if "=" in pair:
236 k, v = pair.split("=", 1)
237 out[k] = v
238 return out
239
240 for sub in ("Latest", "Update", "Previous"):
241 print(f"\n Request&Type=Events&SubType={sub}")
242 try:
243 out = enc_get(f"Request&Type=Events&SubType={sub}")
244 except Exception as e:
245 print(f" FAILED: {e}")
246 continue
247 if out.startswith("<invalid"):
248 print(f" {out.strip()!r}")
249 continue
250 if out.strip() == "<no response>":
251 print(" <no response> (no events for the selected report / nothing new)")
252 continue
253 if not out.startswith("Event"):
254 print(f" unexpected: {out[:160]!r}")
255 continue
256 kv = parse_kv(out)
257 print(f" LastId = {kv.get('LastId')}")
258 # collect Event0, Event1, ... in order
259 idx = sorted((int(k[5:]) for k in kv if k.startswith("Event") and k[5:].isdigit()))
260 print(f" {len(idx)} event(s) in this reply. First few, decoded:")
261 for n in idx[:8]:
262 val = kv.get(f"Event{n}", "")
263 ts = val[:23] # monitorevent.js: first 23 chars are the timestamp
264 desc = val[23:] # ...the rest is the description sentence
265 typeid = kv.get(f"TypeId{n}", "")
266 user = kv.get(f"User{n}", "")
267 extra = f" [TypeId={typeid}]" if typeid else ""
268 extra += f" [User={user}]" if user and user not in ("", "0") else ""
269 print(f" ts={ts!r} desc={desc!r}{extra}")
270 if len(idx) > 8:
271 print(f" … and {len(idx) - 8} more")
272 if sub == "Latest" and idx:
273 print("\n *** THIS IS YOUR LIVE FEED. ***")
274 print(" Paste the decoded lines above. I'll wire the client to Latest/Update polling and")
275 print(" parse these descriptions (door + user + granted/denied/forced/opened) into the NVR.")
276
277 print("\n" + "=" * 60)
278 print("If Latest shows your events above, we're done exploring - paste it back.")
279 print("If it says <no response>, the selected event report is empty; tell me and I'll add a report picker.")
280 print("=" * 60)
281
282
283if __name__ == "__main__":
284 main()