yaris-obd/deploy/yaris_uplink.py

#!/usr/bin/env python3
"""Manage the Pi's phone-hotspot, home-WiFi, and local-AP priority."""
import json
import os
import subprocess
import sys
import tempfile
import time
import uuid


PHONE_CONNECTION = "phone-hotspot"
HOME_CONNECTION = "home-wifi"
AP_CONNECTION = "Hotspot"
PRIMARY_INTERFACE = "wlan1"    # TP-Link T3U dongle — preferred
FALLBACK_INTERFACE = "wlan0"   # onboard — used if T3U is absent
STATUS_PATH = "/var/lib/yaris-obd/uplink-status.json"
PHONE_KEYFILE = (
    "/etc/NetworkManager/system-connections/"
    "phone-hotspot.nmconnection"
)


class UplinkError(RuntimeError):
    pass


def _interface_exists(iface):
    """Return True if the kernel sees this network interface."""
    return os.path.exists(f"/sys/class/net/{iface}")


def resolve_interface():
    """Return PRIMARY_INTERFACE if it exists, else FALLBACK_INTERFACE."""
    if _interface_exists(PRIMARY_INTERFACE):
        return PRIMARY_INTERFACE
    if _interface_exists(FALLBACK_INTERFACE):
        return FALLBACK_INTERFACE
    return PRIMARY_INTERFACE   # return primary anyway; nmcli will error clearly


# Module-level convenience — resolved once at startup.
INTERFACE = resolve_interface()


def run_nmcli(*args, timeout=45, check=True):
    process = subprocess.run(
        ["nmcli", *args],
        capture_output=True,
        text=True,
        timeout=timeout,
    )
    if check and process.returncode != 0:
        detail = (process.stderr or process.stdout).strip()
        raise UplinkError(detail or f"nmcli exited {process.returncode}")
    return process


def connection_exists(name):
    return run_nmcli(
        "-g",
        "connection.id",
        "connection",
        "show",
        name,
        check=False,
    ).returncode == 0


def active_connection():
    process = run_nmcli(
        "-g",
        "GENERAL.CONNECTION",
        "device",
        "show",
        INTERFACE,
        check=False,
    )
    return process.stdout.strip() if process.returncode == 0 else ""


def ipv4_address():
    process = run_nmcli(
        "-g",
        "IP4.ADDRESS",
        "device",
        "show",
        INTERFACE,
        check=False,
    )
    return (
        process.stdout.strip().split("/", 1)[0]
        if process.returncode == 0
        else ""
    )


def internet_reachable():
    process = run_nmcli(
        "-g",
        "GENERAL.IP4-CONNECTIVITY",
        "device",
        "show",
        INTERFACE,
        check=False,
    )
    return process.returncode == 0 and process.stdout.strip() == "4 (full)"


def phone_ssid():
    process = run_nmcli(
        "-g",
        "802-11-wireless.ssid",
        "connection",
        "show",
        PHONE_CONNECTION,
        check=False,
    )
    return process.stdout.strip() if process.returncode == 0 else ""


def phone_is_visible():
    ssid = phone_ssid()
    if not ssid:
        return False
    process = run_nmcli(
        "-t",
        "--escape",
        "no",
        "-f",
        "SSID",
        "device",
        "wifi",
        "list",
        "ifname",
        INTERFACE,
        check=False,
    )
    return process.returncode == 0 and ssid in process.stdout.splitlines()


def activate(name, wait=30):
    if not connection_exists(name):
        return False
    process = run_nmcli(
        "--wait",
        str(wait),
        "connection",
        "up",
        name,
        timeout=wait + 5,
        check=False,
    )
    return process.returncode == 0


def write_status(action, error=None):
    previous = {}
    try:
        with open(STATUS_PATH, encoding="utf-8") as handle:
            previous = json.load(handle)
    except (FileNotFoundError, OSError, json.JSONDecodeError):
        pass
    current = active_connection()
    last_phone_ip = previous.get("last_phone_ip")
    if current == PHONE_CONNECTION:
        last_phone_ip = ipv4_address() or last_phone_ip
    status = {
        "action": action,
        "active_connection": current,
        "interface": INTERFACE,
        "error": error,
        "last_phone_ip": last_phone_ip,
        "updated_at": time.time(),
    }
    os.makedirs(os.path.dirname(STATUS_PATH), mode=0o755, exist_ok=True)
    temporary_path = None
    try:
        with tempfile.NamedTemporaryFile(
            "w",
            encoding="utf-8",
            dir=os.path.dirname(STATUS_PATH),
            prefix=".uplink-status-",
            delete=False,
        ) as handle:
            temporary_path = handle.name
            os.fchmod(handle.fileno(), 0o644)
            json.dump(status, handle)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary_path, STATUS_PATH)
    except OSError:
        if temporary_path is not None:
            try:
                os.remove(temporary_path)
            except FileNotFoundError:
                pass
        raise


def restore_fallback():
    """Try home-wifi on wlan1. wlan0 AP is permanent and never touched."""
    if activate(HOME_CONNECTION):
        write_status("connected_home")
        return True
    write_status("no_connection", "No home WiFi or phone hotspot available.")
    return True


def _ap_client_count():
    """Return number of stations connected to the AP on wlan0."""
    try:
        proc = subprocess.run(
            ["iw", "dev", "wlan0", "station", "dump"],
            capture_output=True, text=True, timeout=5,
        )
        return proc.stdout.count("Station ")
    except (OSError, subprocess.SubprocessError):
        return 0


def check_uplink():
    """Manage internet uplink on wlan1 only. wlan0 is a permanent AP, never touched."""
    current = active_connection()
    phone_configured = connection_exists(PHONE_CONNECTION)

    if current == PHONE_CONNECTION:
        if internet_reachable():
            write_status("phone_connected")
            return True
        if activate(HOME_CONNECTION):
            write_status("phone_without_internet_connected_home")
            return True
        if active_connection() == PHONE_CONNECTION or activate(PHONE_CONNECTION):
            write_status(
                "phone_connected_without_internet",
                "Phone hotspot is connected but internet is not reachable.",
            )
            return True
        return _connect_home_or_idle()

    if phone_configured:
        # Only take down the phone hotspot if it's up — never touch the AP on wlan0
        was_phone = current == PHONE_CONNECTION
        if was_phone:
            run_nmcli("connection", "down", PHONE_CONNECTION, check=False)
        run_nmcli("device", "wifi", "rescan", "ifname", INTERFACE, check=False)
        time.sleep(5)
        if not phone_is_visible():
            if not was_phone and active_connection():
                write_status("phone_not_in_range_current_connection_preserved")
                return True
            return _connect_home_or_idle()
        if activate(PHONE_CONNECTION):
            time.sleep(3)
            if internet_reachable():
                write_status("connected_phone")
                return True
            if activate(HOME_CONNECTION):
                write_status("phone_without_internet_connected_home")
                return True
            if active_connection() == PHONE_CONNECTION:
                write_status(
                    "phone_connected_without_internet",
                    "Phone hotspot is connected but internet is not reachable.",
                )
                return True
            return _connect_home_or_idle()
        if active_connection() == HOME_CONNECTION:
            write_status("phone_unavailable_home_preserved")
            return True
        return _connect_home_or_idle()

    # Phone not configured. Home wifi is the only uplink on wlan1.
    if current == HOME_CONNECTION:
        write_status("home_connected_no_phone")
        return True

    # Not on home-wifi — try to connect
    return _connect_home_or_idle()


def _connect_home_or_idle():
    """Try home-wifi on wlan1. wlan0 AP is never touched."""
    if activate(HOME_CONNECTION):
        write_status("connected_home")
        return True
    write_status("no_uplink_available", "No home WiFi or phone hotspot in range.")
    return True


def wait_for_phone(timeout=120):
    if not connection_exists(PHONE_CONNECTION):
        raise UplinkError("No phone hotspot is configured.")
    if active_connection() == PHONE_CONNECTION:
        write_status("phone_connected")
        return True

    write_status("waiting_for_phone")
    time.sleep(10)

    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        run_nmcli("device", "wifi", "rescan", "ifname", INTERFACE, check=False)
        time.sleep(5)
        if phone_is_visible():
            return check_uplink()

    restored = _connect_home_or_idle()
    if restored:
        write_status(
            "phone_wait_timed_out_fallback_restored",
            "Phone hotspot did not appear within two minutes.",
        )
    return restored


def validate_credentials(payload):
    if not isinstance(payload, dict):
        raise UplinkError("Expected a JSON object.")
    ssid = payload.get("ssid")
    password = payload.get("password")
    if not isinstance(ssid, str) or not ssid or len(ssid.encode("utf-8")) > 32:
        raise UplinkError("SSID must be between 1 and 32 bytes.")
    if "\x00" in ssid or "\n" in ssid or "\r" in ssid:
        raise UplinkError("SSID contains unsupported characters.")
    if ";" in ssid:
        raise UplinkError("SSID cannot contain a semicolon.")
    if not isinstance(password, str) or not 8 <= len(password.encode("utf-8")) <= 63:
        raise UplinkError("Password must be between 8 and 63 bytes.")
    if "\x00" in password or "\n" in password or "\r" in password:
        raise UplinkError("Password contains unsupported characters.")
    return ssid, password


def keyfile_escape(value):
    leading_spaces = len(value) - len(value.lstrip(" "))
    trailing_spaces = len(value) - len(value.rstrip(" "))
    end = len(value) - trailing_spaces if trailing_spaces else len(value)
    middle = value[leading_spaces:end].replace("\\", "\\\\")
    return "\\s" * leading_spaces + middle + "\\s" * trailing_spaces


def configure():
    raw = sys.stdin.buffer.read(2049)
    if len(raw) > 2048:
        raise UplinkError("Configuration request is too large.")
    try:
        payload = json.loads(raw)
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise UplinkError("Invalid configuration JSON.") from error
    ssid, password = validate_credentials(payload)

    was_active = active_connection() == PHONE_CONNECTION
    run_nmcli("connection", "delete", PHONE_CONNECTION, check=False)
    keyfile = f"""[connection]
id={PHONE_CONNECTION}
uuid={uuid.uuid4()}
type=wifi
interface-name={INTERFACE}
autoconnect=true
autoconnect-priority=20

[wifi]
mode=infrastructure
ssid={keyfile_escape(ssid)}
mtu=1400

[wifi-security]
key-mgmt=wpa-psk
psk={keyfile_escape(password)}

[ipv4]
method=auto
dhcp-send-hostname=true
dhcp-hostname=yaris
dhcp-client-id=yaris-obd

[ipv6]
method=auto
"""
    temporary_path = None
    try:
        os.makedirs(os.path.dirname(PHONE_KEYFILE), mode=0o700, exist_ok=True)
        with tempfile.NamedTemporaryFile(
            "w",
            encoding="utf-8",
            dir=os.path.dirname(PHONE_KEYFILE),
            prefix=".phone-hotspot-",
            delete=False,
        ) as handle:
            temporary_path = handle.name
            os.fchmod(handle.fileno(), 0o600)
            handle.write(keyfile)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary_path, PHONE_KEYFILE)
        temporary_path = None
        run_nmcli("connection", "reload")
        if not connection_exists(PHONE_CONNECTION):
            raise UplinkError("NetworkManager did not load the phone hotspot.")
    except (OSError, UplinkError, subprocess.TimeoutExpired):
        if temporary_path is not None:
            try:
                os.remove(temporary_path)
            except FileNotFoundError:
                pass
        try:
            os.remove(PHONE_KEYFILE)
        except FileNotFoundError:
            pass
        run_nmcli("connection", "reload", check=False)
        run_nmcli("connection", "delete", PHONE_CONNECTION, check=False)
        if was_active:
            try:
                restore_fallback()
            except (OSError, subprocess.TimeoutExpired, UplinkError):
                pass
        raise
    write_status("phone_configured")
    if was_active:
        check_uplink()


def remove():
    was_active = active_connection() == PHONE_CONNECTION
    run_nmcli("connection", "delete", PHONE_CONNECTION, check=False)
    try:
        os.remove(PHONE_KEYFILE)
    except FileNotFoundError:
        pass
    if was_active:
        restore_fallback()
    else:
        write_status("phone_removed")


def record_current_connection():
    current = active_connection()
    actions = {
        PHONE_CONNECTION: "network_phone_connected",
        HOME_CONNECTION: "network_home_connected",
        AP_CONNECTION: "network_local_hotspot_active",
    }
    write_status(actions.get(current, "network_disconnected"))


def main():
    if len(sys.argv) != 2 or sys.argv[1] not in {
        "check",
        "configure",
        "connect",
        "record",
        "wait",
        "remove",
    }:
        print(
            "Usage: yaris-uplink {check|configure|connect|record|wait|remove}",
            file=sys.stderr,
        )
        return 2
    action = sys.argv[1]
    try:
        if action == "configure":
            configure()
        elif action == "remove":
            remove()
        elif action == "wait":
            wait_for_phone()
        elif action == "record":
            record_current_connection()
        else:
            check_uplink()
    except (OSError, subprocess.TimeoutExpired, UplinkError) as error:
        try:
            write_status(f"{action}_failed", str(error))
        except OSError:
            pass
        print(str(error), file=sys.stderr)
        return 1
    return 0


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