seismo-relay/bridges/mm_probe.py

#!/usr/bin/env python3
"""
mm_probe.py — answer "why can't we reach this unit?" in one command.

THOR reports a failed connection as "disconnected" and nothing else.  That single
word covers at least four completely different faults with four different fixes,
and telling them apart is the difference between a modem reboot and a site visit:

  * **connection refused**   something answered and said no — wrong port, or the
                             modem is refusing a further session
  * **connect timed out**    nothing answered at all — trusted-IP whitelist,
                             firewall, or the modem is off the network
  * **connected, no reply**  the MODEM answered but the unit did not.  The TCP
                             path is fine; the modem is not forwarding to serial.
                             This is the signature of a wedged transparent-TCP
                             session, and it is the one THOR cannot distinguish
                             from any of the others
  * **replied**              the unit is alive; the problem is upstream software

Read-only.  It sends `POLL`, then optionally `SERIAL` and the state read — the
same three commands THOR's own connection check uses — and never writes.

Usage
-----
    python3 bridges/mm_probe.py 63.45.161.30:9034
    python3 bridges/mm_probe.py 10.0.0.8:12345 --timeout 5
    python3 bridges/mm_probe.py <host:port> --slots 3

`--slots N` opens N connections at once and reports how many the far end accepts.
A transparent-TCP modem typically serves **one** session; if the first succeeds
and the rest are refused or hang, that confirms the single-slot behaviour and
explains why a leaked session takes a unit offline until the slot frees.

Works for both series: a Series III reply opens `DLE STX`, a Micromate reply
opens with a bare `STX`, so the probe also tells you which one answered.
"""

from __future__ import annotations

import argparse
import socket
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from minimateplus.framing import build_bw_frame  # noqa: E402

DLE, STX, ETX = 0x10, 0x02, 0x03


def destuff(raw: bytes) -> bytes:
    """Strip framing and DLE escapes; return the payload without its checksum."""
    i = 1 if raw and raw[0] == STX else (2 if len(raw) > 1 and raw[1] == STX else 0)
    out = bytearray()
    while i < len(raw):
        b = raw[i]
        if b == DLE and i + 1 < len(raw):
            out.append(raw[i + 1])
            i += 2
            continue
        if b == ETX:
            break
        out.append(b)
        i += 1
    return bytes(out[:-1]) if len(out) > 1 else b""


# Reads are two-step on Series III: a probe at offset 0, then a data read at the
# block's length.  THOR sends these offsets, and they also work on a Micromate.
OFFSETS = {0x5B: 0x0030, 0x15: 0x000A, 0x49: 0xFFFF}


def exchange(sock: socket.socket, sub: int, timeout: float) -> tuple[bytes, float]:
    sock.sendall(build_bw_frame(sub, OFFSETS.get(sub, 0)))
    t0 = time.time()
    buf, deadline = b"", t0 + timeout
    sock.settimeout(0.3)
    while time.time() < deadline:
        try:
            chunk = sock.recv(4096)
            if not chunk:
                break
            buf += chunk
            if buf.endswith(bytes([ETX])) and len(buf) > 8:
                break
        except TimeoutError:
            continue
        except OSError:
            break
    return buf, time.time() - t0


def step(n: int, label: str, result: str) -> None:
    print(f"  [{n}] {label:.<28} {result}")


def probe(host: str, port: int, timeout: float) -> int:
    print(f"\ntarget {host}:{port}   (read-only: POLL, SERIAL, state)\n")

    # ── 1. TCP ────────────────────────────────────────────────────────────
    t0 = time.time()
    try:
        sock = socket.create_connection((host, port), timeout=timeout)
    except ConnectionRefusedError:
        step(1, "TCP connect", f"REFUSED after {1000*(time.time()-t0):.0f} ms")
        print("\nverdict: something answered and actively refused.")
        print("  Not a silent firewall drop — the host is reachable.")
        print("  Wrong port, the service is down, or the modem is refusing")
        print("  an additional session because its one slot is in use.")
        return 2
    except (TimeoutError, socket.timeout):
        step(1, "TCP connect", f"TIMED OUT after {time.time()-t0:.1f} s")
        print("\nverdict: nothing answered at all.")
        print("  A silent drop, which is what a trusted-IP whitelist looks")
        print("  like — it discards rather than refuses.  Check the modem's")
        print("  Trusted IPs (and note a VPN changes the IP you arrive from),")
        print("  the firewall, and whether the modem is on the network.")
        return 3
    except OSError as e:
        step(1, "TCP connect", f"FAILED: {e}")
        return 4
    step(1, "TCP connect", f"ok ({1000*(time.time()-t0):.0f} ms)")

    # ── 2. POLL ───────────────────────────────────────────────────────────
    try:
        raw, dt = exchange(sock, 0x5B, timeout)
    except OSError as e:
        step(2, "POLL", f"send failed: {e}")
        sock.close()
        return 4

    if not raw:
        step(2, "POLL", f"NO REPLY in {timeout:.1f} s")
        print("\nverdict: the MODEM answered but the unit did not.")
        print("  TCP is fine end to end — something accepted the connection.")
        print("  What is missing is the serial side.  Most likely the modem is")
        print("  not forwarding to its serial port, which is what a wedged")
        print("  transparent-TCP session looks like: the slot is held by a")
        print("  connection that never closed.")
        print("\n  Try, in order:")
        print("    1. ACEmanager -> TCP Idle Timeout.  If 0/disabled, a stale")
        print("       session holds the slot forever.  2 minutes is the value")
        print("       this project standardised on.")
        print("    2. Reboot the modem.  If that fixes it, the modem was")
        print("       holding state and the timeout is the permanent fix.")
        print("    3. Check the unit's own screen — serial cable, power.")
        sock.close()
        return 5

    series = "Series III (DLE STX)" if raw[0] == DLE else "Micromate (bare STX)"
    step(2, "POLL", f"reply {len(raw)} B in {1000*dt:.0f} ms")
    p = destuff(raw)
    ok = len(p) > 3 and p[2] == 0xFF - 0x5B
    step(3, "frame", f"{'valid' if ok else 'MALFORMED'}, {series}")
    if not ok:
        print("\nverdict: something replied, but not a seismograph.")
        print("  Another service is on this port, or the modem is in a mode")
        print("  that injects its own text (check Quiet Mode / AT echo).")
        print(f"  first bytes: {raw[:16].hex(' ')}")
        sock.close()
        return 6

    # ── 3. identity + state ───────────────────────────────────────────────
    for n, (sub, label) in enumerate(((0x15, "serial"), (0x49, "state")), start=4):
        try:
            r, dt = exchange(sock, sub, timeout)
            d = destuff(r)[5:]
            if sub == 0x15:
                # serial is a null-terminated run; a further field follows it
                serial = bytes(d[11:]).split(b"\x00")[0]
                step(n, label, serial.decode("ascii", "replace") or "(empty)")
            else:
                step(n, label, "MONITORING" if len(d) > 11 and d[11] else "idle")
        except OSError:
            step(n, label, "no reply")

    sock.close()
    print("\nverdict: the unit is alive and answering.")
    print("  If THOR still shows it disconnected, the fault is in THOR, not")
    print("  the network or the device.")
    return 0


def slots(host: str, port: int, n: int, timeout: float) -> None:
    print(f"\nopening {n} simultaneous connections to {host}:{port}\n")
    held = []
    for i in range(n):
        try:
            s = socket.create_connection((host, port), timeout=timeout)
            held.append(s)
            step(i + 1, f"connection {i+1}", "accepted")
        except ConnectionRefusedError:
            step(i + 1, f"connection {i+1}", "REFUSED")
        except (TimeoutError, socket.timeout):
            step(i + 1, f"connection {i+1}", "timed out")
        except OSError as e:
            step(i + 1, f"connection {i+1}", f"failed: {e}")
    print(f"\n{len(held)} of {n} accepted.")
    if len(held) == 1:
        print("  Single-slot behaviour confirmed — this far end serves ONE")
        print("  session at a time.  A connection that is never closed takes")
        print("  the unit offline until the idle timeout frees the slot.")
    for s in held:
        s.close()


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("target", help="host:port, e.g. 63.45.161.30:9034")
    ap.add_argument("--timeout", type=float, default=10.0)
    ap.add_argument("--slots", type=int, metavar="N",
                    help="open N simultaneous connections to test single-slot behaviour")
    a = ap.parse_args()
    host, _, port = a.target.rpartition(":")
    if not host:
        ap.error("target must be host:port")
    if a.slots:
        slots(host, int(port), a.slots, a.timeout)
        return 0
    return probe(host, int(port), a.timeout)


if __name__ == "__main__":
    raise SystemExit(main())