from __future__ import annotations import logging import time from dataclasses import dataclass from .config import AppConfig from .portal import CaptivePortalLogin, PortalLoginError from .windows import ( WifiCommandError, connect_wifi, disconnect_wifi, ensure_hardware_mac, get_wifi_status, randomize_mac, request_wifi_scan, restart_adapter, set_interface_enabled, set_wlan_autoconfig, test_connectivity, wait_for_ssid, ) @dataclass(slots=True) class ServiceState: disconnected_since_monotonic: float | None = None last_connect_attempt_monotonic: float | None = None last_recovery_monotonic: float | None = None last_mac_refresh_monotonic: float | None = None startup_policy_pending: bool = True class WifiBackgroundService: def __init__(self, config: AppConfig, logger: logging.Logger) -> None: self.config = config self.logger = logger self.state = ServiceState( startup_policy_pending=(config.randomize_mac_on_start and config.allow_unsafe_mac_changes), ) self.portal_login = CaptivePortalLogin(config.portal, config.selenium, logger) def run_forever(self, run_once: bool = False) -> int: if self.config.active_network.randomize_mac and not self.config.allow_unsafe_mac_changes: self.logger.warning( "MAC randomization is configured for '%s' but disabled because " "allow_unsafe_mac_changes=false", self.config.active_network.ssid, ) self.logger.info( "Starting OSU Wi-Fi monitor for network '%s' (%s)", self.config.active_network.key, self.config.active_network.ssid, ) while True: try: self._run_iteration() except WifiCommandError as exc: if run_once: self.logger.error("%s", exc) return 1 self.logger.error("Monitor iteration failed: %s", exc) except Exception: if run_once: self.logger.exception("Monitor iteration failed unexpectedly") return 1 self.logger.exception("Monitor iteration failed unexpectedly") if run_once: return 0 time.sleep(self.config.monitor.check_interval_seconds) def _run_iteration(self) -> None: network = self.config.active_network status = get_wifi_status(self.config.wifi_interface_name) now = time.monotonic() on_target_network = status.is_connected and status.ssid == network.ssid internet_ok = on_target_network and test_connectivity(self.config.connectivity_checks) if self._mac_randomization_enabled() and self.state.startup_policy_pending: self.logger.info("Applying startup MAC policy for %s", network.ssid) self._recover_connection(reason="startup MAC policy") self.state.startup_policy_pending = False self.state.disconnected_since_monotonic = None return if self._mac_randomization_enabled() and self._mac_refresh_due(): self.logger.info("MAC refresh interval reached for %s", network.ssid) self._recover_connection(reason="scheduled MAC refresh", refresh_mac=True) self.state.disconnected_since_monotonic = None return if not on_target_network and self._soft_connect_due(now): if self._attempt_soft_connect(status): self.state.disconnected_since_monotonic = None return if on_target_network and internet_ok: self.state.disconnected_since_monotonic = None return if on_target_network and network.requires_portal_login and not internet_ok: self.logger.info("Connected to %s without internet; attempting portal login", network.ssid) self._login_if_needed() self.state.disconnected_since_monotonic = None return now = time.monotonic() if self.state.disconnected_since_monotonic is None: self.state.disconnected_since_monotonic = now self.logger.warning( "Wi-Fi is unhealthy. interface=%s state=%s ssid=%s", status.interface_name, status.state, status.ssid, ) return unhealthy_seconds = now - self.state.disconnected_since_monotonic if unhealthy_seconds < self.config.monitor.disconnect_grace_seconds: return if self.state.last_recovery_monotonic is not None: cooldown = now - self.state.last_recovery_monotonic if cooldown < self.config.monitor.adapter_reset_cooldown_seconds: return self._recover_connection(reason=f"connection unhealthy for {int(unhealthy_seconds)} seconds") self.state.disconnected_since_monotonic = None def _recover_connection(self, reason: str, refresh_mac: bool | None = None) -> None: network = self.config.active_network refresh_mac = self._mac_randomization_enabled() if refresh_mac is None else refresh_mac self.logger.warning("Starting recovery: %s", reason) status = get_wifi_status(self.config.wifi_interface_name) interface_name = status.interface_name or self.config.wifi_interface_name if not interface_name: raise WifiCommandError("Could not determine the Wi-Fi interface name") adapter_locator = status.description or interface_name if network.restore_hardware_mac and self.config.allow_unsafe_mac_changes: try: restored = ensure_hardware_mac(adapter_locator) if restored: self.logger.info("Restored hardware MAC before reconnecting to %s", network.ssid) except WifiCommandError as exc: self.logger.warning("Could not restore hardware MAC: %s", exc) if refresh_mac: mac_address = randomize_mac(adapter_locator) self.state.last_mac_refresh_monotonic = time.monotonic() self.logger.info("Randomized MAC to %s", mac_address) self._prepare_interface_for_connect(interface_name) try: disconnect_wifi(interface_name) except WifiCommandError as exc: self.logger.warning("Wi-Fi disconnect before recovery failed, continuing: %s", exc) try: self.logger.info("Restarting Wi-Fi adapter %s", interface_name) restart_adapter(interface_name) except WifiCommandError as exc: self.logger.warning( "Adapter restart failed, continuing with direct reconnect attempt: %s", exc, ) time.sleep(self.config.monitor.reconnect_wait_seconds) self.logger.info("Reconnecting to %s after recovery actions", network.ssid) connect_wifi( network.profile_name, interface_name, ssid=network.ssid, auto_create_open_profile=network.auto_create_open_profile, ) status = wait_for_ssid( network.ssid, self.config.monitor.connection_timeout_seconds, interface_name, ) self.state.last_recovery_monotonic = time.monotonic() if not status.is_connected or status.ssid != network.ssid: raise WifiCommandError( f"Failed to reconnect to {network.ssid}; final status: {self._format_status(status)}", ) self.logger.info("Reconnected to %s on interface %s", status.ssid, interface_name) if network.requires_portal_login: self._login_if_needed() def _attempt_soft_connect(self, status) -> bool: network = self.config.active_network interface_name = status.interface_name or self.config.wifi_interface_name if not interface_name: raise WifiCommandError( "Could not determine the Wi-Fi interface name. Set wifi_interface_name in config.yaml.", ) self.logger.info( "Attempting direct Wi-Fi connection to %s on interface %s", network.ssid, interface_name, ) self.state.last_connect_attempt_monotonic = time.monotonic() self._prepare_interface_for_connect(interface_name) connect_wifi( network.profile_name, interface_name, ssid=network.ssid, auto_create_open_profile=network.auto_create_open_profile, ) new_status = wait_for_ssid( network.ssid, self.config.monitor.connection_timeout_seconds, interface_name, ) if not new_status.is_connected or new_status.ssid != network.ssid: self.logger.warning( "Direct Wi-Fi connection to %s did not succeed; final status: %s", network.ssid, self._format_status(new_status), ) return False self.logger.info("Connected to %s without adapter reset", network.ssid) if network.requires_portal_login: self._login_if_needed() return True def _login_if_needed(self) -> None: try: if not self.portal_login.login_if_present(): if test_connectivity(self.config.connectivity_checks): self.logger.info("Captive portal was not detected and connectivity checks passed") return raise WifiCommandError( "Connected to the Wi-Fi network, but internet is unavailable and the captive portal " "login page was not detected", ) except PortalLoginError as exc: raise WifiCommandError(f"Captive portal login failed: {exc}") from exc deadline = time.monotonic() + self.config.monitor.connection_timeout_seconds while time.monotonic() < deadline: if test_connectivity(self.config.connectivity_checks): self.logger.info("Captive portal login succeeded") return self.logger.info("Waiting for internet access after captive portal submission") time.sleep(2) raise WifiCommandError("Captive portal login completed but internet is still unavailable") def _mac_refresh_due(self) -> bool: network = self.config.active_network if not self._mac_randomization_enabled() or not network.mac_refresh_hours: return False if self.state.last_mac_refresh_monotonic is None: return False refresh_seconds = network.mac_refresh_hours * 3600 return (time.monotonic() - self.state.last_mac_refresh_monotonic) >= refresh_seconds def _mac_randomization_enabled(self) -> bool: network = self.config.active_network return self.config.allow_unsafe_mac_changes and network.randomize_mac def _soft_connect_due(self, now: float) -> bool: if self.state.last_connect_attempt_monotonic is None: return True return ( now - self.state.last_connect_attempt_monotonic ) >= self.config.monitor.connect_retry_cooldown_seconds def _prepare_interface_for_connect(self, interface_name: str) -> None: for action_name, action in ( ("enable interface", lambda: set_interface_enabled(interface_name, True)), ("enable WLAN autoconfig", lambda: set_wlan_autoconfig(interface_name, True)), ("scan Wi-Fi networks", lambda: request_wifi_scan(interface_name)), ): try: action() except WifiCommandError as exc: self.logger.warning("Could not %s for %s: %s", action_name, interface_name, exc) @staticmethod def _format_status(status) -> str: return f"interface={status.interface_name} state={status.state} ssid={status.ssid}"