#!/usr/bin/env python3
"""W-0 REACHABILITY PREFLIGHT for releases#304 W2 / #297. NOT RUN until doyle names it.

One question, three possible answers, no cargo and no product code:

    can kitsubito open a TCP connection to a listener on THIS Windows box,
    on the port the bootstrap listener would use?

  ADMITTED                -- the connect succeeds and reads the sentinel.
  BLOCKED-BY-HOST-FIREWALL-- the connect fails AND this box's firewall has no
                             rule admitting the port (measured, not assumed).
  BLOCKED-BY-ACL          -- the connect fails while the host firewall DOES
                             admit the port -> the remaining layer is the
                             tailnet ACL, which is the operator question on
                             releases#304 Q2 and which this script never edits.

The two failures must not share a face: a rig that cannot tell them apart
sends the operator to the wrong dashboard. That is the whole point of the
`netsh` read -- without it a failed connect is one silence with two causes.

TWO SUBJECTS, and the script picks by MEASUREMENT rather than by assumption:

  (a) A REAL bootstrap listener is already up on this box (this node has been
      LAN-EXPOSED on 5470 all session). Then the faithful probe is a GET of
      /install on THAT listener from the peer -- which is literally the #297
      field report, not a simulation of it. The script only READS it: it never
      stops, restarts or reconfigures an operator-started listener.
  (b) Nothing is listening. Then a throwaway python socket answers the narrower
      question "is this port reachable from there".

Case (a) is strictly better evidence and also the reason the original shape of
this script was wrong: it would have tried to bind 5470, collided with the live
listener, and reported INVALID-PROBE on a box whose answer was sitting right
there. Either way the product's rule-WRITING is untested here on purpose --
that is W-1/W-2, and using the unbuilt product in its own gate is circular.

Run from the lane worktree on HFENDULEAM:
    python w0-preflight.py --port 5470 --peer reavus@kitsubito --self-ip <tailnet v4>

NOTE the tailnet IPv4 is NOT the 192.168.1.81 the LAN banner prints -- that is
the LAN interface. Pass the address the PEER would use, measured on the peer.
"""

import argparse
import json
import socket
import subprocess
import threading
import time

SENTINEL = b"spt-w0-preflight\n"


def run(cmd, **kw):
    """Direct capture, never through a pipe: `$?` after a pipeline is the
    LAST command's status, which has already fabricated one confident green
    in this milestone (hertz, 2026-09-11)."""
    p = subprocess.run(cmd, capture_output=True, text=True, **kw)
    return p.returncode, p.stdout, p.stderr


def firewall_admits(port):
    """Three-value, like the W2 F17 gate's precondition read: True / False /
    None(unanswerable). None is NOT False -- a netsh that did not run tells us
    nothing about the rule, and reporting that as 'no rule' would invent a
    measurement."""
    rc, out, err = run(["netsh", "advfirewall", "firewall", "show", "rule",
                        "name=all", "dir=in", "protocol=tcp", "verbose"])
    if rc != 0:
        return None, f"netsh exit {rc}: {err.strip()[:200]}"
    blocks = out.split("\n\n")
    hits = [b for b in blocks
            if f"LocalPort:" in b and str(port) in b and "Enabled:" in b
            and "Yes" in b and "Allow" in b]
    return (len(hits) > 0), f"{len(hits)} enabled allow rule(s) naming port {port}"


def serve_once(port, stop):
    srv = socket.socket()
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind(("0.0.0.0", port))
    srv.listen(1)
    srv.settimeout(1.0)
    while not stop.is_set():
        try:
            conn, _ = srv.accept()
        except socket.timeout:
            continue
        conn.sendall(SENTINEL)
        conn.close()
    srv.close()


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--port", type=int, default=5470)
    ap.add_argument("--peer", default="reavus@kitsubito")
    ap.add_argument("--self-ip", required=True,
                    help="this box's tailnet IPv4 as the PEER sees it -- passed in, "
                         "never guessed from a local interface list")
    ap.add_argument("--wait", type=float, default=8.0)
    args = ap.parse_args()

    result = {"port": args.port, "peer": args.peer, "self_ip": args.self_ip,
              "started_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}

    admits, how = firewall_admits(args.port)
    result["host_firewall_admits"] = admits
    result["host_firewall_evidence"] = how

    # WHICH SUBJECT? Measure, do not assume: a live listener on this port is
    # the better subject AND makes binding impossible, so the same read decides
    # both questions.
    try:
        with socket.create_connection(("127.0.0.1", args.port), timeout=3):
            pass
        occupied = True
    except OSError:
        occupied = False
    result["subject"] = "live-bootstrap-listener" if occupied else "throwaway-socket"

    stop = threading.Event()
    if not occupied:
        t = threading.Thread(target=serve_once, args=(args.port, stop), daemon=True)
        t.start()
        time.sleep(0.5)

    # Positive control FIRST: if the LOOPBACK fetch fails, whatever is (or is
    # not) on that port is not answering, and every remote verdict below would
    # be about my own socket rather than about the network. A probe that cannot
    # prove itself alive proves nothing.
    probe_path = "/install" if occupied else ""
    try:
        with socket.create_connection(("127.0.0.1", args.port), timeout=3) as s:
            if occupied:
                s.sendall(f"GET {probe_path} HTTP/1.0\r\nHost: 127.0.0.1\r\n\r\n".encode())
                local_ok = b"200" in s.recv(256)
            else:
                local_ok = s.recv(64) == SENTINEL
    except OSError as e:
        local_ok = False
        result["local_error"] = str(e)
    result["positive_control_loopback"] = local_ok

    if not local_ok:
        result["verdict"] = "INVALID-PROBE"
        stop.set()
        print(json.dumps(result, indent=2))
        return 2

    if occupied:
        # -sS: no progress bar, errors still spoken. -o /dev/null: the BYTES are
        # W-2's claim, not W-0's -- here only "did anything answer" is asked.
        remote = (f"curl -sS -m {int(args.wait)} -o /dev/null "
                  f"-w '%{{http_code}}' http://{args.self_ip}:{args.port}/install")
    else:
        remote = (f"timeout {int(args.wait)} bash -c "
                  f"'exec 3<>/dev/tcp/{args.self_ip}/{args.port} && head -c 64 <&3'")
    rc, out, err = run(["ssh", "-o", "BatchMode=yes",
                        "-o", f"ConnectTimeout={int(args.wait)}", args.peer, remote])
    stop.set()
    result["remote_cmd"] = remote
    result["remote_exit"] = rc
    result["remote_stdout"] = out.strip()[:200]
    result["remote_stderr"] = err.strip()[:300]

    reached = (rc == 0 and ("200" in out if occupied
                            else SENTINEL.decode().strip() in out))
    if reached:
        result["verdict"] = "ADMITTED"
    elif admits is None:
        result["verdict"] = "BLOCKED-LAYER-UNMEASURED"   # a labelled hole, not a cause
    elif admits:
        result["verdict"] = "BLOCKED-BY-ACL"
    else:
        result["verdict"] = "BLOCKED-BY-HOST-FIREWALL"

    print(json.dumps(result, indent=2))
    return 0 if reached else 1


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