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``).

Credential derivation (confirmed from the disassembly, 20-char SN rule)::

    ssid = "AP_" + sn[11:]          pwd = sn[8:]
    else (invalid length)
    ssid = "AP_" + sn               pwd = "12345678"

Actual joining is delegated to a :class:`WifiBackend` so this module stays
testable without a radio.  On a Linux desktop/laptop use
:class:`NmcliWifiBackend` (NetworkManager); in a container or when joining
manually use :class:`NoOpWifiBackend`.
"""

from __future__ import annotations

import logging
import subprocess
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import 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"


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

    ssid: str
    password: str
    gateway_ip: str = DEVICE_GATEWAY_IP


def derive_hotspot_credentials(sn: str) -> "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``).
    """
    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)


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

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

    @abstractmethod
    def leave(self) -> None:
        """Disconnect from the device hotspot."""

    @abstractmethod
    def get_ip(self) -> Optional[str]:
        """Our own IP on the current Wi-Fi network (device-side gateway view)."""

    def is_connected(self, profile: HotspotProfile) -> bool:
        """Best-effort: did the join stick?"""
        return False


class NoOpWifiBackend(WifiBackend):
    """Assumes the user has already connected (manually or via another tool).

    Useful in sandboxes/containers without a radio, and when the host's own
    tooling handles the association.
    """

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

    def join(self, profile: HotspotProfile) -> None:
        self._profile = profile
        self.joined = True

    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."""

    def __init__(self, nmcli: str = "nmcli", ipcmd: str = "ip") -> None:
        self._nmcli = nmcli
        self._ipcmd = ipcmd

    def _run(self, *args: str) -> str:
        proc = subprocess.run(
            [self._nmcli, *args],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=30,
        )
        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 join(self, profile: HotspotProfile) -> None:
        self._run("device", "wifi", "connect", profile.ssid,
                  "password", profile.password)

    def leave(self) -> None:
        self._run("device", "wifi", "disconnect")

    def get_ip(self) -> Optional[str]:
        try:
            out = self._run("-g", "IP4.ADDRESS", "device", "show")
        except (RuntimeError, subprocess.TimeoutExpired, OSError):
            return None
        for line in out.splitlines():
            addr = line.strip()
            if addr.startswith("10.") or addr.startswith("192.168."):
                return addr
        return None

    def is_connected(self, profile: HotspotProfile) -> bool:
        try:
            out = self._run("-g", "SSID", "device", "wifi", "show")
        except (RuntimeError, subprocess.TimeoutExpired, OSError):
            return False
        return profile.ssid in out.splitlines()


class WifiManager:
    """Facade over a :class:`WifiBackend` (keeps the original stub API)."""

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

    # -- new, clean API ----------------------------------------------------
    def join_device_hotspot(self, sn: str) -> HotspotProfile:
        """Derive the profile from the SN and join.  Returns the profile."""
        profile = derive_hotspot_credentials(sn)
        self.backend.join(profile)
        self.connected = True
        return profile

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

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

    # -- original stub names (kept for API compatibility) -------------------
    def query_connect_wifi(self, ssid: str, password: Optional[str] = None,
                           sn: Optional[str] = None):
        """``queryConnectWifi`` — join (or rejoin) a network."""
        if sn is not None and password is None:
            profile = derive_hotspot_credentials(sn)
        else:
            profile = HotspotProfile(ssid=ssid, password=password or "")
        self.backend.join(profile)
        self.connected = True
        return profile

    def wifi_onboarding_add_wifi(self, ssid: str, password: str):
        """``wifiOnboardingAddWifi`` — register a known network."""
        self.backend.join(HotspotProfile(ssid=ssid, password=password))
        return True

    def get_wifi_signal_strength(self) -> Optional[int]:
        if not isinstance(self.backend, NmcliWifiBackend):
            return None
        try:
            out = self.backend._run("-g", "SIGNAL", "device", "wifi", "show")
        except (RuntimeError, subprocess.TimeoutExpired, OSError):
            return None
        for line in out.splitlines():
            digits = "".join(ch for ch in line if ch.isdigit())
            if digits:
                return int(digits)
        return None