Initial commit

tools/gpon_telnet.py

@@ -0,0 +1,71 @@
+#!/usr/bin/env python3
+"""Minimal scriptable telnet client for probing the GPON stick CLI.
+Usage: gpon_telnet.py [--host H] [--port P] [--user U] [--pass P] [--wait S] [cmd ...]
+Host/port/user/pass default to the .env settings (see ../example.env).
+Each positional cmd is sent as one line after login; output captured per cmd.
+Reads extra commands (one per line) from stdin if --stdin given.
+"""
+import socket, sys, time, re
+import gpon_env
+
+args = sys.argv[1:]
+HOST, PORT = gpon_env.HOST, gpon_env.PORT
+USER, PASS, WAIT, USE_STDIN = gpon_env.USER, gpon_env.PASS, 2.5, False
+cmds = []
+while args:
+    a = args.pop(0)
+    if a == '--host': HOST = args.pop(0)
+    elif a == '--port': PORT = int(args.pop(0))
+    elif a == '--user': USER = args.pop(0)
+    elif a == '--pass': PASS = args.pop(0)
+    elif a == '--wait': WAIT = float(args.pop(0))
+    elif a == '--stdin': USE_STDIN = True
+    else: cmds.append(a)
+if USE_STDIN:
+    cmds += [l.rstrip('\n') for l in sys.stdin if l.strip() != '']
+
+def strip_iac(data, outbuf):
+    res = bytearray(); i = 0
+    while i < len(data):
+        b = data[i]
+        if b == 0xFF and i+1 < len(data):
+            cmd = data[i+1]
+            if cmd in (251,252,253,254) and i+2 < len(data):
+                opt = data[i+2]
+                if cmd == 253: outbuf.append(bytes([255,252,opt]))   # DO->WONT
+                elif cmd == 251: outbuf.append(bytes([255,254,opt]))  # WILL->DONT
+                i += 3; continue
+            elif cmd == 250:  # SB..SE
+                j = data.find(b'\xff\xf0', i+2)
+                i = len(data) if j == -1 else j+2; continue
+            else:
+                i += 2; continue
+        res.append(b); i += 1
+    return bytes(res)
+
+s = socket.create_connection((HOST, PORT), timeout=8)
+
+def rx(timeout):
+    s.settimeout(0.6); buf = bytearray(); end = time.time()+timeout
+    while time.time() < end:
+        try:
+            data = s.recv(4096)
+            if not data: break
+            ob = []
+            buf.extend(strip_iac(data, ob))
+            for o in ob: s.sendall(o)
+        except socket.timeout:
+            if buf: break
+    return buf.decode('latin-1','replace')
+
+out = rx(3.0)
+if re.search(r'(ogin|sername)', out):
+    s.sendall(USER.encode()+b'\r\n'); out += rx(2.0)
+if re.search(r'assword', out, re.I):
+    s.sendall(PASS.encode()+b'\r\n'); out += rx(3.0)
+print("===== BANNER/LOGIN =====\n"+out)
+for c in cmds:
+    s.sendall(c.encode()+b'\r\n'); o = rx(WAIT)
+    print("\n===== CMD: %r =====\n%s" % (c, o))
+try: s.close()
+except Exception: pass