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.  Two quite different")
        print("  causes produce this, and they are NOT distinguishable from")
        print("  here:")
        print("\n  1. SOMEONE ELSE HOLDS THE SESSION.  These modems bridge ONE")
        print("     TCP session to serial at a time.  A second connection is")
        print("     accepted and then simply not forwarded.  Confirmed 2026-09-26:")
        print("     with THOR connected this probe saw exactly this; the moment")
        print("     THOR disconnected the same probe returned the serial number.")
        print("     ** Check whether THOR (or anything else) has the unit first. **")
        print("\n  2. The serial path is genuinely broken — a stale session the")
        print("     modem never released, a cable the unit cannot enumerate, or")
        print("     a unit that is off.")
        print("\n  Try, in order:")
        print("    1. Disconnect any other client and re-probe.  If it answers,")
        print("       it was contention, not a fault.")
        print("    2. The cable's chipset.  A Micromate drives FTDI and CDC-ACM")
        print("       only — a Prolific PL2303 gives it no serial port at all.")
        print("       lsusb: FTDI is 0403, Prolific 067b.")
        print("    3. Power-cycle the UNIT with the cable attached (hold power")
        print("       5 s, through the two-stage prompt).  Its USB host rescans")
        print("       on cold boot; it may not on hot-swap.")
        print("    4. AirLink OS -> TCP Idle Timeout.  If 0/disabled, a stale")
        print("       session holds the slot indefinitely.  2 minutes is the")
        print("       value this project standardised on.")
        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())