RH821GWV-DG bridge conifguration

hathway_bridge_setup.py

#!/usr/bin/env python3
"""
isp_bridge_setup.py -- one-shot configurator for the MediaTek/Airoha RH821GWV-DG
GPON HGU (TrendChip/Airoha firmware, busybox v1.00, Boa web server).

Applies, idempotently and persistently:
  0.  Enable SSH over the web UI (HTTP) -- so a fresh/reset unit (SSH OFF by
      default) can be driven; then everything below runs over SSH.
  1.  Admin password (web + SSH) via tcapi (commit auto-regenerates /etc/passwd)
  2.  Disable the login CAPTCHA            (/boaroot overlay: index2.asp)
  3.  Unlock the ISP-locked WAN page       (/boaroot overlay: net-wanset.asp LockTR69Node)
  4.  Phantom-WAN watchdog                 (/vendor/phantom_guard.sh + boa wrapper)
  5.  WAN -> Bridge mode, VLAN tagged
  5b. Bind the bridge to a LAN port        (the port cabled to your downstream PPPoE router)

The device has no Python interpreter, so this script is a *driver*: it ships
busybox-sh primitives to the router and runs them in a single SSH session (so the
mid-run password change does not break the connection). Persistence: /vendor
(mtdblock, jffs2) survives reboots and /usr/script/vendor_load.sh re-mounts the
/boaroot + /userfs overlays on every boot.

No secrets are stored in this file. Everything comes from environment variables:
  ISP_HOST       router mgmt IP           (default: auto-detect the LAN gateway,
                                            else 192.168.200.1)
  ISP_PORT       ssh port                 (default 22)
  ISP_USER       login user               (default user -- the secondary account, UID 0)
  ISP_PASS       CURRENT login password   (default 1234 -- that account's vendor default)
  ISP_NEW_PASS   new admin password       (required)
  ISP_VLAN       WAN VLAN id              (default 202)
  ISP_BIND_PORT  LAN1|LAN2|LAN3|LAN4      (default LAN1)
  NO_BRIDGE      "1" skips steps 5 and 5b (default 0)
  ISP_HTTP_PORT  Boa web UI port          (default 80; used only for step 0)
  ENABLE_SSH     "0" skips the web SSH-enable bootstrap (default 1)

Requirements: python3 (standard library only for the SSH-enable bootstrap), the
system `ssh`, and `sshpass` (the firmware's dropbear 0.52 only does password auth
with legacy crypto). The password is passed to sshpass via the environment (never
argv), so it does not appear in the process list. The SSH-enable step uses HTTP
Basic auth; the web login CAPTCHA is client-side only, so it works on a stock unit.

Usage:
  ISP_PASS='...' ISP_NEW_PASS='...' python3 isp_bridge_setup.py
  python3 isp_bridge_setup.py --dry-run     # print the remote payload, connect to nothing

Reliability note: steps 1-4 are verified. Steps 5/5b (bridge flip + LAN bind) are
built from the exact attribute values read off a working bridged unit but were not
test-written live (that would drop a working link); the script reads each back and
warns if it did not take. Reliable fallback: the WAN page is unlocked by step 3, so
you can set Mode=Bridge and tick the LAN bind in the web UI by hand.
"""

import os
import re
import sys
import time
import shlex
import socket
import base64
import platform
import subprocess
import urllib.error
import urllib.parse
import urllib.request


def detect_gateway():
    """Best-effort default-gateway IPv4 -- the router itself when this script is
    run from a machine on the router's LAN. Cross-platform, no third-party deps.
    Returns the dotted-quad string, or None if it can't be determined."""
    system = platform.system()
    if system == "Linux":
        cmds = [["ip", "-4", "route", "show", "default"]]
    elif system == "Windows":
        cmds = [["powershell", "-NoProfile", "-Command",
                 "(Get-NetRoute -DestinationPrefix 0.0.0.0/0 | "
                 "Sort-Object RouteMetric | Select-Object -First 1).NextHop"]]
    else:  # macOS / *BSD
        cmds = [["route", "-n", "get", "default"], ["netstat", "-rn", "-f", "inet"]]
    patterns = [
        r"default via (\d+\.\d+\.\d+\.\d+)",     # ip route
        r"gateway:\s*(\d+\.\d+\.\d+\.\d+)",       # route -n get default
        r"^default\s+(\d+\.\d+\.\d+\.\d+)",       # netstat -rn
        r"^\s*(\d+\.\d+\.\d+\.\d+)\s*$",           # powershell NextHop
    ]
    for cmd in cmds:
        try:
            out = subprocess.run(cmd, capture_output=True, text=True, timeout=5).stdout
        except Exception:  # noqa: BLE001 - detection is best-effort
            continue
        for pat in patterns:
            m = re.search(pat, out, re.MULTILINE)
            if m:
                return m.group(1)
    return None

# --------------------------------------------------------------------------- #
# Configuration (all from the environment -- no secrets or PII in this file)   #
# --------------------------------------------------------------------------- #
# Router mgmt IP: explicit ISP_HOST wins; else auto-detect the default gateway
# (the router, when run from its LAN); else fall back to the vendor default.
HOST = os.environ.get("ISP_HOST") or detect_gateway() or "192.168.200.1"
PORT = os.environ.get("ISP_PORT", "22")
USER = os.environ.get("ISP_USER", "user")
CUR_PASS = os.environ.get("ISP_PASS", "1234")  # secondary-account vendor default (UID 0 / root; can
                                               # enable SSH over HTTP and run tcapi); override per unit
NEW_PASS = os.environ.get("ISP_NEW_PASS")
VLAN = os.environ.get("ISP_VLAN", "202")
BIND_PORT = os.environ.get("ISP_BIND_PORT", "LAN1")
NO_BRIDGE = os.environ.get("NO_BRIDGE", "0") == "1"
HTTP_PORT = os.environ.get("ISP_HTTP_PORT", "80")   # Boa web UI port (for the SSH-enable bootstrap)
ENABLE_SSH = os.environ.get("ENABLE_SSH", "1") != "0"  # HTTP-enable SSH first (needed on a fresh unit)

# Legacy crypto so OpenSSH will talk to the router's dropbear 0.52.
SSH_OPTS = [
    "-p", str(PORT),
    "-o", "StrictHostKeyChecking=no",
    "-o", "UserKnownHostsFile=/dev/null",
    "-o", "HostKeyAlgorithms=+ssh-rsa",
    "-o", "PubkeyAcceptedAlgorithms=+ssh-rsa",
    "-o", "KexAlgorithms=+diffie-hellman-group1-sha1",
    "-o", "Ciphers=+aes128-cbc",
    "-o", "ConnectTimeout=10",
]

# --------------------------------------------------------------------------- #
# The on-router payload. The top few variables are injected by Python; the body #
# below is plain busybox sh and runs entirely on the device.                   #
# --------------------------------------------------------------------------- #
SHELL_BODY = r'''
HAL=/userfs/bin/sample_hal_wan
TCAPI=/userfs/bin/tcapi
log(){ echo "[isp-setup] $*"; }

# ------------------------------------------------------------------ STEP 1 ---
log "STEP 1/5: admin password (web + SSH)"
CUR=$($TCAPI get Account_Entry web_passwd 2>/dev/null)
if [ "$CUR" = "$NEWPW" ]; then
  log "  already set, skipping"
else
  $TCAPI set Account_Entry web_passwd "$NEWPW"
  $TCAPI commit Account_Entry     # commit also regenerates /etc/passwd (empty-salt MD5)
  $TCAPI save
  log "  web_passwd updated (SSH + web now use the new password)"
fi

# --------------------------------------------------------------- STEP 2 + 3 ---
log "STEP 2+3/5: /boaroot overlay (disable captcha + unlock WAN page)"
# Build the overlay upperdir from the PRISTINE lower pages. If the overlay is
# already mounted, unmount first so /boaroot exposes the read-only originals.
mkdir -p /vendor/boaroot/upper/cgi-bin /vendor/boaroot/work
if mount | grep -q "overlay on /boaroot"; then
  log "  unmounting existing /boaroot overlay to read pristine lower"
  umount /boaroot 2>/dev/null
fi
SRC_C=/boaroot/cgi-bin/index2.asp
SRC_W=/boaroot/cgi-bin/net-wanset.asp
DST_C=/vendor/boaroot/upper/cgi-bin/index2.asp
DST_W=/vendor/boaroot/upper/cgi-bin/net-wanset.asp
if [ ! -s "$SRC_C" ] || [ ! -s "$SRC_W" ]; then
  log "  ERROR: source pages missing ($SRC_C / $SRC_W) - skipping /boaroot overlay"
else
  # captcha: neutralize the ASP gates (isCaptchaSupported = "Yes" -> "DISABLED")
  sed 's#"isCaptchaSupported", "h") = "Yes"#"isCaptchaSupported", "h") = "DISABLED"#g' "$SRC_C" > "$DST_C"
  # WAN unlock: the single TR069 lock call -> the enable branch
  sed 's/LockTR69Node(1);/LockTR69Node(0);/' "$SRC_W" > "$DST_W"
  CAP_Y=$(grep -c 'isCaptchaSupported", "h") = "Yes"' "$DST_C")   # want 0
  LK1=$(grep -c 'LockTR69Node(1);' "$DST_W")                       # want 0
  SC=$(grep -c '' "$SRC_C"); DC=$(grep -c '' "$DST_C")             # line counts must match
  SW=$(grep -c '' "$SRC_W"); DW=$(grep -c '' "$DST_W")             # (sed sub preserves lines)
  if [ "$CAP_Y" = "0" ] && [ "$LK1" = "0" ] && [ "$SC" = "$DC" ] && [ "$SW" = "$DW" ]; then
    mount -t overlay overlay -o ro,lowerdir=/boaroot,upperdir=/vendor/boaroot/upper,workdir=/vendor/boaroot/work /boaroot
    log "  overlay mounted: captcha OFF, WAN page UNLOCKED (index2 $DC lines, net-wanset $DW lines)"
  else
    log "  ERROR transform/verify failed (capY=$CAP_Y lock1=$LK1 lines c:$SC/$DC w:$SW/$DW) - NOT mounting"
    rm -f "$DST_C" "$DST_W"
  fi
fi

# ------------------------------------------------------------------ STEP 4 ---
log "STEP 4/5: phantom-WAN watchdog"
# If a Bridge WAN on VLAN $VLAN and a Route WAN on VLAN $VLAN both exist (the OLT
# re-pushes the Route one on reprovision), delete the phantom Route WAN.
cat > /vendor/phantom_guard.sh <<PGUARD
#!/bin/sh
LOG=/tmp/phantom_guard.log
VLAN=$VLAN
HAL=$HAL
[ -f /tmp/.pguard_running ] && exit 0
echo running > /tmp/.pguard_running
echo "\$(awk '{print \$1; exit}' /proc/uptime) phantom_guard started" >> \$LOG
sleep 30
while true; do
  paths=\$(\$HAL list_wan_path 2>/dev/null | awk -F= '/info.path/{print \$NF}')
  has_bridge=0; route_targets=""
  for p in \$paths; do
    pp=\$(echo "\$p" | sed 's/\.\$//')
    mode=\$(\$HAL get_wan_attr "\$pp" 1 2>/dev/null | sed -n 's/.*value = \([^,]*\),.*/\1/p')
    vlan=\$(\$HAL get_wan_attr "\$pp" 5 2>/dev/null | sed -n 's/.*value = \([^,]*\),.*/\1/p')
    [ "\$mode" = "Bridge" ] && [ "\$vlan" = "\$VLAN" ] && has_bridge=1
    [ "\$mode" = "Route" ] && [ "\$vlan" = "\$VLAN" ] && route_targets="\$route_targets \$pp"
  done
  if [ "\$has_bridge" = "1" ] && [ -n "\$route_targets" ]; then
    for rp in \$route_targets; do
      \$HAL del_link "\$rp" >/dev/null 2>&1
      echo "\$(awk '{print \$1; exit}' /proc/uptime) removed phantom Route \$rp vlan \$VLAN" >> \$LOG
    done
  fi
  sleep 15
done
PGUARD
chmod +x /vendor/phantom_guard.sh
# Auto-start hook: overlay a wrapper over /userfs/bin/boa that launches the guard
# then execs the real boa. Preserve the genuine binary as boa.real, and never
# re-wrap the wrapper (copy only when boa.real is missing / too small).
mkdir -p /vendor/userfs/upper/bin /vendor/userfs/work
RS=$(ls -l /vendor/userfs/upper/bin/boa.real 2>/dev/null | awk '{print $5+0}')
[ -z "$RS" ] && RS=0
if [ "$RS" -lt 100000 ]; then
  cp /userfs/bin/boa /vendor/userfs/upper/bin/boa.real
  log "  preserved real boa -> boa.real ($(ls -l /vendor/userfs/upper/bin/boa.real | awk '{print $5}') bytes)"
else
  log "  boa.real already preserved ($RS bytes), skipping copy"
fi
cat > /vendor/userfs/upper/bin/boa <<'BOAW'
#!/bin/sh
# Launch the phantom-WAN watchdog once, then run the real web server unchanged.
/vendor/phantom_guard.sh >/dev/null 2>&1 &
exec /userfs/bin/boa.real "$@"
BOAW
chmod +x /vendor/userfs/upper/bin/boa
if ! mount | grep -q "overlay on /userfs"; then
  mount -t overlay overlay -o ro,lowerdir=/userfs,upperdir=/vendor/userfs/upper,workdir=/vendor/userfs/work /userfs
  log "  /userfs overlay mounted (boa wrapper wired for next boa start / reboot)"
else
  log "  /userfs overlay already mounted (boa wrapper in place)"
fi
# Start the watchdog now too (self-locks via /tmp/.pguard_running -> no double run).
/vendor/phantom_guard.sh >/dev/null 2>&1 &
log "  watchdog launched (pid $!)"

# -------------------------------------------------------------- STEP 5 + 5b ---
log "STEP 5/5: WAN -> BRIDGE mode, VLAN $VLAN tagged, bound to $BINDPORT"
if [ "${NO_BRIDGE:-0}" = "1" ]; then
  log "  NO_BRIDGE=1 set, skipping bridge flip + bind"
else
  # HAL attribute map (sample_hal_wan get/set_wan_attr <path> <n>):
  #   1=WanMode 3=ServiceList 4=VLANMode 5=VLANID 12=NAT
  #   17=LAN1 18=LAN2 19=LAN3 20=LAN4  (bridge->LAN bind, Yes/No)
  # Values below were read off a working bridged unit (WanMode=Bridge,
  # ServiceList=INTERNET, VLANMode=TAG, attr17=Yes => bound to LAN1).
  case "$BINDPORT" in
    LAN1) BIND_ATTR=17 ;; LAN2) BIND_ATTR=18 ;; LAN3) BIND_ATTR=19 ;; LAN4) BIND_ATTR=20 ;;
    *) BIND_ATTR=17 ;;
  esac
  FOUND=""
  for p in $($HAL list_wan_path 2>/dev/null | awk -F= '/info.path/{print $NF}'); do
    pp=$(echo "$p" | sed 's/\.$//')
    vlan=$($HAL get_wan_attr "$pp" 5 2>/dev/null | sed -n 's/.*value = \([^,]*\),.*/\1/p')
    mode=$($HAL get_wan_attr "$pp" 1 2>/dev/null | sed -n 's/.*value = \([^,]*\),.*/\1/p')
    [ "$vlan" != "$VLAN" ] && continue
    FOUND="$pp"
    log "  WAN path $pp: mode=$mode vlan=$vlan"
    if [ "$mode" != "Bridge" ]; then
      $HAL set_wan_attr "$pp" 4 TAG        # VLANMode    = TAG
      $HAL set_wan_attr "$pp" 5 "$VLAN"    # VLANID
      $HAL set_wan_attr "$pp" 3 INTERNET   # ServiceList = INTERNET (was TR069_INTERNET in Route)
      $HAL set_wan_attr "$pp" 1 Bridge     # WanMode     = Route -> Bridge
      $HAL effect_link "$pp"               # apply: tear down PPPoE, bring up the bridge
      NOW=$($HAL get_wan_attr "$pp" 1 2>/dev/null | sed -n 's/.*value = \([^,]*\),.*/\1/p')
      log "  WanMode now reads: $NOW"
      [ "$NOW" != "Bridge" ] && log "  WARNING bridge flip did not take via CLI - set Mode=Bridge in the unlocked web UI"
    else
      log "  already Bridge -> not re-flipping"
    fi
    # Bind the bridge to $BINDPORT (this is what lets bridged VLAN $VLAN traffic
    # egress on the port cabled to the downstream PPPoE router).
    for a in 17 18 19 20; do
      if [ "$a" = "$BIND_ATTR" ]; then $HAL set_wan_attr "$pp" "$a" Yes 2>/dev/null
      else $HAL set_wan_attr "$pp" "$a" No 2>/dev/null; fi
    done
    $HAL effect_link "$pp" 2>/dev/null
    BS=$($HAL get_wan_attr "$pp" "$BIND_ATTR" 2>/dev/null | sed -n 's/.*value = \([^,]*\),.*/\1/p')
    log "  bound bridge -> $BINDPORT (attr$BIND_ATTR reads: $BS)"
    [ "$BS" != "Yes" ] && log "  WARNING bind did not take via CLI - tick $BINDPORT in the web UI Bind section"
  done
  $TCAPI save 2>/dev/null                  # persist WAN changes to flash
  [ -z "$FOUND" ] && log "  WARNING: no VLAN-$VLAN WAN path present yet (OLT not provisioned?); flip via web UI once it appears"
fi

log "DONE. password changed; captcha off; WAN page unlocked; watchdog installed."
log "Reboot-safe: /vendor overlays re-mount via vendor_load.sh."
'''


def http_enable_ssh():
    """Enable SSH via the Boa web UI (no SSH needed) so a fresh/reset unit can be
    driven. Mirrors the hidden /cgi-bin/telnet.asp SSH form:
        POST ssh_flag=1 & sshEnable_flag=Yes  -> tcapi SSH_Entry.Enable=Yes + commit
    Boa auth = HTTP Basic + UID/PSW cookies + a SESSIONID from a prior GET (the
    login CAPTCHA is client-side only, so this works even before step 2 disables it).
    Uses only the standard library. Returns the POST HTTP status.
    """
    base = f"http://{HOST}:{HTTP_PORT}"
    auth = "Basic " + base64.b64encode(f"{USER}:{CUR_PASS}".encode()).decode()
    # 1. GET the page to obtain a SESSIONID cookie (a 401 still sets it).
    try:
        resp = urllib.request.urlopen(
            urllib.request.Request(base + "/cgi-bin/telnet.asp", headers={"Authorization": auth}),
            timeout=10)
        set_cookies = resp.headers.get_all("Set-Cookie") or []
    except urllib.error.HTTPError as e:
        set_cookies = e.headers.get_all("Set-Cookie") or []
    sid = next((c.split(";", 1)[0] for c in set_cookies if c.startswith("SESSIONID=")), None)
    cookie = "; ".join(x for x in [sid, f"UID={USER}", f"PSW={CUR_PASS}"] if x)
    # 2. POST the SSH-enable form.
    data = urllib.parse.urlencode({"ssh_flag": "1", "sshEnable_flag": "Yes"}).encode()
    req = urllib.request.Request(
        base + "/cgi-bin/telnet.asp", data=data,
        headers={"Authorization": auth, "Cookie": cookie,
                 "Referer": base + "/cgi-bin/telnet.asp",
                 "Content-Type": "application/x-www-form-urlencoded"})
    return urllib.request.urlopen(req, timeout=10).status


def wait_for_ssh(timeout=20):
    """Poll the SSH port until dropbear is up after enabling it (or timeout)."""
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            with socket.create_connection((HOST, int(PORT)), timeout=3):
                return True
        except OSError:
            time.sleep(2)
    return False


def build_payload():
    """Prepend the Python-supplied parameters to the busybox-sh body."""
    header = (
        "set -u\n"
        f"NEWPW={shlex.quote(NEW_PASS or '')}\n"
        f"VLAN={shlex.quote(str(VLAN))}\n"
        f"BINDPORT={shlex.quote(BIND_PORT)}\n"
        f"NO_BRIDGE={'1' if NO_BRIDGE else '0'}\n"
    )
    return header + SHELL_BODY


def die(msg, code=2):
    sys.stderr.write(f"error: {msg}\n")
    sys.exit(code)


def main():
    dry_run = "--dry-run" in sys.argv[1:]
    payload = build_payload()

    if dry_run:
        # Redact the injected password before printing the payload.
        redacted = payload.replace(shlex.quote(NEW_PASS or ""), "'<NEW_ADMIN_PASSWORD>'", 1)
        sys.stdout.write(redacted)
        return 0

    if BIND_PORT not in ("LAN1", "LAN2", "LAN3", "LAN4"):
        die(f"ISP_BIND_PORT must be LAN1..LAN4 (got {BIND_PORT!r})")
    if not CUR_PASS:
        die("ISP_PASS (current login password) is required")
    if not NEW_PASS:
        die("ISP_NEW_PASS (new admin password to set) is required")

    # Step 0: enable SSH over the web UI first (fresh/reset units have it OFF).
    if ENABLE_SSH:
        try:
            st = http_enable_ssh()
            print(f"[driver] SSH enabled via web UI (HTTP {st}); waiting for dropbear ...",
                  flush=True)
            if not wait_for_ssh():
                print("[driver] WARNING: SSH port not open yet; trying anyway.", flush=True)
        except Exception as e:  # noqa: BLE001 - best-effort bootstrap
            print(f"[driver] WARNING: HTTP SSH-enable failed ({e}); "
                  f"assuming SSH is already enabled and continuing.", flush=True)

    cmd = ["sshpass", "-e", "ssh"] + SSH_OPTS + [f"{USER}@{HOST}", "sh -s"]
    env = dict(os.environ, SSHPASS=CUR_PASS)

    print(f"[driver] connecting to {USER}@{HOST}:{PORT} and running setup "
          f"(VLAN={VLAN}, bind={BIND_PORT}, no_bridge={NO_BRIDGE}) ...", flush=True)
    try:
        proc = subprocess.run(cmd, input=payload, env=env, text=True)
    except FileNotFoundError as e:
        die(f"missing dependency: {e.filename} (need `ssh` and `sshpass` on PATH)")
    if proc.returncode != 0:
        die(f"remote run exited {proc.returncode} "
            f"(login/crypto issue, or the password was already changed -- "
            f"set ISP_PASS to the CURRENT password)", proc.returncode)
    print("[driver] done.", flush=True)
    return 0


if __name__ == "__main__":
    sys.exit(main())