franklinwh-agate-client/franklinwh_agate_client/wifi.py

"""Wi-Fi / device-hotspot handling.

Reconstructed from ``franklinwh2/utils/device_wifi_utils.dart`` (Dart side)
and ``com.alternadom.wifiiot.WifiIotPlugin`` (Java side, ``base.apk``) — see
``docs/RE_DART_FINDINGS.md`` §10, §13.

The app joins the **device SoftAP** so that the phone can talk to the aGate
over its local HTTP/socket interfaces (gateway ``10.100.1.1``).

How this module maps onto that::

    * credential derivation (``derive_hotspot_credentials``) — faithful
      reconstruction of ``getDeviceApSsid`` / ``getDeviceApDefaultPwd`` (§10).
    * the association itself (``WifiBackend.join`` / ``leave``) — platform
      work.  :class:`NmcliWifiBackend` drives NetworkManager on a Linux
      desktop/laptop; :class:`NoOpWifiBackend` records state only, for
      hosts whose own tooling owns the radio (containers, manual joins).
    * ``WifiManager`` — the facade that makes the join *functional*:
      derive → join → **verify** (poll ``is_connected`` against a deadline,
      fail-closed) → expose our IP and the device gateway.

Note on the app's *other* "wifi" operations: ``connectWifi`` (0x0f) and the
ESP-provisioning/onboarding flow (§4, §9) provision the **device** with a
home router — they are device-side RPCs (see ``LocalDeviceClient`` /
``CommunicationCmd``), not host-radio actions.  The compat shims here
(``query_connect_wifi`` / ``wifi_onboarding_add_wifi``) only perform the
host-side join the app performs at the end of onboarding.
"""

from __future__ import annotations

import logging
import subprocess
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Callable, List, Optional

log = logging.getLogger(__name__)

#: The aGate SoftAP gateway the phone talks to after joining.
DEVICE_GATEWAY_IP = "10.100.1.1"

#: SN length that triggers the "valid" derivation branch (sn.length == 20).
VALID_SN_LENGTH = 20

FALLBACK_PASSWORD = "12345678"

#: Address prefixes tried (in order) when picking the host address after a
#: join.  The aGate gateway is 10.100.1.1 and its DHCP normally hands out
#: 10.100.1.x; the other prefixes are generic private-range fallbacks.
_ADDRESS_PREFIX_PREFERENCE = ("10.100.", "10.", "192.168.", "172.16.")


@dataclass(frozen=True)
class HotspotProfile:
    """A device's hotspot connection parameters."""

    ssid: str
    password: str
    gateway_ip: str = DEVICE_GATEWAY_IP


class HotspotJoinError(RuntimeError):
    """A join was issued but the association never verified in time."""


def derive_hotspot_credentials(sn: str,
                               gateway_ip: Optional[str] = None) -> "HotspotProfile":
    """``getDeviceApSsid`` / ``getDeviceApDefaultPwd``.

    Offline fallback: for a *20-character* SN the SSID is ``"AP_" + sn[11:]``
    and the password is ``sn[8:]``; anything else yields ``"AP_" + sn`` /
    ``"12345678"`` (the app logs ``"deviceSn is invalid"`` in that case).

    The authoritative values come from the live device
    (``/terminal/getHotSpotInfo`` + ``/terminal/getHosSpotPassword``).

    ``gateway_ip`` overrides the default SoftAP gateway (``10.100.1.1``) —
    e.g. when the aGate is reached on its home-LAN address instead.
    """
    if len(sn) == VALID_SN_LENGTH:
        ssid = "AP_" + sn[11:]
        password = sn[8:]
    else:
        log.warning("deviceSn is invalid (len=%d, expected %d): using full-SN fallback",
                    len(sn), VALID_SN_LENGTH)
        ssid = "AP_" + sn
        password = FALLBACK_PASSWORD
    return HotspotProfile(ssid=ssid, password=password,
                          gateway_ip=gateway_ip or DEVICE_GATEWAY_IP)


class WifiBackend(ABC):
    """Platform abstraction for joining/leaving a Wi-Fi network.

    Contract:

    * ``join`` must raise on hard failure (missing tool, rejected SSID…).
    * ``leave`` is best-effort: do not raise when there is simply nothing
      to disconnect.
    * ``is_connected`` must be **fail-closed** (return ``False`` when
      uncertain, e.g. the underlying tooling is absent).  The manager polls
      it to decide whether a join actually stuck, so a wrong ``True`` is
      far worse than a late ``False``.
    """

    @abstractmethod
    def join(self, profile: HotspotProfile) -> None:
        """Connect to ``profile``.  Raises on failure."""

    @abstractmethod
    def leave(self) -> None:
        """Disconnect from the current Wi-Fi network (best-effort)."""

    @abstractmethod
    def get_ip(self) -> Optional[str]:
        """Our own IPv4 address on the currently associated network."""

    def is_connected(self, profile: HotspotProfile) -> bool:
        """Fail-closed: is ``profile`` the network we are associated to?"""
        return False

    def get_signal(self) -> Optional[int]:
        """Signal strength (0-100) of the associated AP, or ``None``."""
        return None


class NoOpWifiBackend(WifiBackend):
    """Records join/leave state without touching a radio.

    For sandboxes/containers and for hosts where the operator joins the
    device hotspot manually (or via the host's own tooling).  Verification
    then trivially passes: the backend records the profile ``join`` was
    given.  Use :meth:`assume_connected` when the association was made
    outside of this process.
    """

    def __init__(self, local_ip: Optional[str] = None) -> None:
        self._local_ip = local_ip
        self._profile: Optional[HotspotProfile] = None
        self.joined: bool = False

    def assume_connected(self, profile: HotspotProfile,
                         local_ip: Optional[str] = None) -> None:
        """Declare that the host is already associated with ``profile``."""
        if local_ip is not None:
            self._local_ip = local_ip
        self._profile = profile
        self.joined = True

    def join(self, profile: HotspotProfile) -> None:
        self.assume_connected(profile)

    def leave(self) -> None:
        self._profile = None
        self.joined = False

    def get_ip(self) -> Optional[str]:
        return self._local_ip

    def is_connected(self, profile: HotspotProfile) -> bool:
        return self.joined and self._profile == profile


class NmcliWifiBackend(WifiBackend):
    """Join via ``nmcli`` (NetworkManager) — for Linux desktops/laptops.

    Verified against nmcli/NetworkManager 1.52 source (``devices.c``):

    * association state: ``nmcli -t -f SSID,ACTIVE device wifi list`` —
      the ``ACTIVE`` column is ``yes`` *only* for the AP we are associated
      to.  (There is no ``device wifi show`` subcommand; the command table
      is ``list / connect / hotspot / rescan / show-password`` and a bare
      ``show`` is accepted only by prefix-matching ``show-password``.  Using
      a scan list or that accident to mean "associated" is wrong, and
      version-dependent — hence ``-f SSID,ACTIVE``.)
    * our IP: ``nmcli -g IP4.ADDRESS device show <wifi-iface>`` per Wi-Fi
      interface (never the unfiltered all-device dump, which would happily
      report a wired/VPN address).
    * signal: the ``SIGNAL`` column of the active AP from the same list.
    """

    def __init__(self, nmcli: str = "nmcli", timeout: float = 30.0) -> None:
        self._nmcli = nmcli
        self._timeout = timeout

    def _run(self, *args: str) -> str:
        proc = subprocess.run(
            [self._nmcli, *args],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=self._timeout,
        )
        if proc.returncode != 0:
            raise RuntimeError(
                f"{' '.join(args)} failed: "
                f"{proc.stderr.decode('utf-8', 'replace').strip()}"
            )
        return proc.stdout.decode("utf-8", "replace").strip()

    def _wifi_ifaces(self) -> List[str]:
        """Wi-Fi interface names in an associated/active state."""
        try:
            out = self._run("-t", "-f", "DEVICE,TYPE,STATE", "device", "show")
        except (RuntimeError, subprocess.TimeoutExpired, OSError):
            return []
        ifaces: List[str] = []
        for line in out.splitlines():
            parts = line.split(":")
            if len(parts) >= 3 and parts[1] == "wifi" and parts[2] in ("activated", "connected"):
                ifaces.append(parts[0])
        return ifaces

    def join(self, profile: HotspotProfile) -> None:
        self._run("device", "wifi", "connect", profile.ssid,
                  "password", profile.password)

    def leave(self) -> None:
        # Best-effort: there may be nothing to disconnect (radio off, no
        # association).  Tolerate it — teardown must not fail the flow.
        try:
            self._run("device", "wifi", "disconnect")
        except (RuntimeError, subprocess.TimeoutExpired, OSError) as exc:
            log.warning("wifi disconnect did not complete cleanly: %s", exc)

    def get_ip(self) -> Optional[str]:
        candidates: List[str] = []
        for dev in self._wifi_ifaces():
            try:
                out = self._run("-g", "IP4.ADDRESS", "device", "show", dev)
            except (RuntimeError, subprocess.TimeoutExpired, OSError):
                continue
            candidates.extend(a.strip() for a in out.splitlines() if a.strip())
        for prefix in _ADDRESS_PREFIX_PREFERENCE:
            for addr in candidates:
                if addr.startswith(prefix):
                    return addr
        return candidates[0] if candidates else None

    def is_connected(self, profile: HotspotProfile) -> bool:
        try:
            out = self._run("-t", "-f", "SSID,ACTIVE", "device", "wifi", "list")
        except (RuntimeError, subprocess.TimeoutExpired, OSError):
            return False  # fail-closed: cannot confirm association
        return f"{profile.ssid}:yes" in out.splitlines()

    def get_signal(self) -> Optional[int]:
        try:
            out = self._run("-t", "-f", "SSID,SIGNAL,ACTIVE", "device", "wifi", "list")
        except (RuntimeError, subprocess.TimeoutExpired, OSError):
            return None
        for line in out.splitlines():
            # SSIDs may contain ':'; the trailing field is the ACTIVE flag.
            parts = line.rsplit(":", 2)
            if len(parts) == 3 and parts[2] == "yes":
                try:
                    return int(parts[1])
                except ValueError:
                    return None
        return None


class WifiManager:
    """Facade over a :class:`WifiBackend` — derive, join, **verify**.

    ``join_*`` do not take the backend's exit code on faith: after
    ``join`` they poll ``backend.is_connected(profile)`` against
    ``settle_timeout`` (NetworkManager can report success before the
    association actually settles, and it can fail a few seconds later).
    An unverified join raises :class:`HotspotJoinError` and leaves
    ``connected`` ``False``.  ``verify=False`` restores fire-and-forget.
    """

    def __init__(self, backend: Optional[WifiBackend] = None) -> None:
        self.backend = backend or NoOpWifiBackend()
        self.connected = False
        self.last_profile: Optional[HotspotProfile] = None

    # -- internal ----------------------------------------------------------
    def _join(self, profile: HotspotProfile, *, verify: bool,
              settle_timeout: float, poll_interval: float) -> HotspotProfile:
        self.last_profile = profile
        self.connected = False
        self.backend.join(profile)  # raises on hard failure
        ok = self._verify(
            lambda: self.backend.is_connected(profile),
            settle_timeout, poll_interval,
        ) if verify else True
        self.connected = ok
        if not ok:
            raise HotspotJoinError(
                f"join to '{profile.ssid}' was issued but association was "
                f"not confirmed within {settle_timeout:g}s"
            )
        return profile

    @staticmethod
    def _verify(check: Callable[[], bool], timeout: float,
                poll_interval: float) -> bool:
        """Poll ``check()`` until it is true or ``timeout`` elapses."""
        deadline = time.monotonic() + timeout
        while True:
            try:
                if check():
                    return True
            except Exception as exc:  # a broken backend must fail closed
                log.debug("is_connected check raised: %s", exc)
            if time.monotonic() >= deadline:
                return False
            time.sleep(max(poll_interval, 0.0))

    # -- device hotspot ------------------------------------------------------
    def join_device_hotspot(self, sn: str, *, verify: bool = True,
                            settle_timeout: float = 30.0,
                            poll_interval: float = 1.0,
                            gateway_ip: Optional[str] = None) -> HotspotProfile:
        """Derive the profile from the SN, join, and verify the association.

        ``gateway_ip`` overrides the device address recorded on the
        profile (default ``10.100.1.1``) without changing the radio join.
        """
        profile = derive_hotspot_credentials(sn, gateway_ip=gateway_ip)
        return self._join(profile, verify=verify,
                          settle_timeout=settle_timeout,
                          poll_interval=poll_interval)

    def join_network(self, ssid: str, password: str, *, verify: bool = True,
                     settle_timeout: float = 30.0,
                     poll_interval: float = 1.0) -> HotspotProfile:
        """Join an arbitrary network (host side)."""
        return self._join(HotspotProfile(ssid=ssid, password=password),
                          verify=verify, settle_timeout=settle_timeout,
                          poll_interval=poll_interval)

    def leave_device_hotspot(self) -> None:
        self.backend.leave()
        self.connected = False
        self.last_profile = None

    # -- status --------------------------------------------------------------
    def local_ip(self) -> Optional[str]:
        """Our own IP on the associated network (``None`` if unknown)."""
        return self.backend.get_ip()

    def gateway_ip(self) -> str:
        """The aGate's IP on the hotspot (where the local API lives)."""
        return self.last_profile.gateway_ip if self.last_profile else DEVICE_GATEWAY_IP

    def get_wifi_signal_strength(self) -> Optional[int]:
        return self.backend.get_signal()

    # -- original stub names (kept for API compatibility) -------------------
    def query_connect_wifi(self, ssid: str, password: Optional[str] = None,
                           sn: Optional[str] = None, **join_kwargs: object):
        """Join an arbitrary network (host side); with ``sn`` only, join the
        device hotspot — same as :meth:`join_device_hotspot`.

        NB: in the app, ``queryConnectWifi`` is the *device* RPC
        (``CommunicationCmd.CONNECT_WIFI``, 0x0f) that tells the aGate to
        join a router — a provisioning operation, not a host-radio action.
        """
        if sn is not None and password is None:
            return self.join_device_hotspot(sn, **join_kwargs)
        return self.join_network(ssid, password or "", **join_kwargs)

    def wifi_onboarding_add_wifi(self, ssid: str, password: str, **join_kwargs: object):
        """Host-side step of the app's Wi-Fi onboarding: after the aGate has
        been provisioned with the home router (device-side — see
        ``CommunicationCmd.SCAN_WIFI_LIST`` / ``CONNECT_WIFI``), the phone
        rejoins the home network.  This performs that host-side rejoin.
        """
        return self.join_network(ssid, password, **join_kwargs)