Improvements

CHANGELOG.md

@@ -2,6 +2,19 @@
 
 All notable changes to this project will be documented in this file.
 
+## [0.2.6] - 2025-08-06
+
+### Added
+
+- Added redaction of data in exceptions when requesting `status()`
+- Additional settings in dataclass (HA Core Issue 150118)
+- Added 'likely' mocked fixture for above issue
+- Added additional devices (see [Contributing](CONTRIBUTE.md) for more information)
+
+### Changed
+
+- Changed name and kwargs for discovery function
+
 ## [0.2.5] - 2025-08-05
 
 ### Added

CONTRIBUTE.md

@@ -7,6 +7,8 @@ It would be very helpful if you would share your configuration data to this proj
 We currently have data on
 
 - Nanostation 5AC (LOCO5AC) - PTP - both AP and Station output of `/status.cgi` present (by @CoMPaTech)
+- Nanobeam 5AC - PTMP - Station (by @PlayFaster)
+- LiteAP GPS - PTMP - AP (by @PlayFaster)
 
 ## Secure your data
 

airos/airos8.py

@@ -10,7 +10,7 @@
 import aiohttp
 from mashumaro.exceptions import InvalidFieldValue, MissingField
 
-from .data import AirOS8Data as AirOSData
+from .data import AirOS8Data as AirOSData, redact_data_smart
 from .exceptions import (
     AirOSConnectionAuthenticationError,
     AirOSConnectionSetupError,
@@ -268,28 +268,44 @@ async def status(self) -> AirOSData:
                 if response.status == 200:
                     try:
                         response_json = json.loads(response_text)
-                        try:
-                            adjusted_json = self.derived_data(response_json)
-                            airos_data = AirOSData.from_dict(adjusted_json)
-                        except (MissingField, InvalidFieldValue) as err:
-                            _LOGGER.exception("Failed to deserialize AirOS data")
-                            raise AirOSKeyDataMissingError from err
-
-                        return airos_data
+                        adjusted_json = self.derived_data(response_json)
+                        airos_data = AirOSData.from_dict(adjusted_json)
+                    except InvalidFieldValue as err:
+                        # Log with .error() as this is a specific, known type of issue
+                        redacted_data = redact_data_smart(response_json)
+                        _LOGGER.error(
+                            "Failed to deserialize AirOS data due to an invalid field value: %s",
+                            redacted_data,
+                        )
+                        raise AirOSKeyDataMissingError from err
+                    except MissingField as err:
+                        # Log with .exception() for a full stack trace
+                        redacted_data = redact_data_smart(response_json)
+                        _LOGGER.exception(
+                            "Failed to deserialize AirOS data due to a missing field: %s",
+                            redacted_data,
+                        )
+                        raise AirOSKeyDataMissingError from err
+
                     except json.JSONDecodeError:
                         _LOGGER.exception(
                             "JSON Decode Error in authenticated status response"
                         )
                         raise AirOSDataMissingError from None
+
+                    return airos_data
                 else:
-                    log = f"Authenticated status.cgi failed: {response.status}. Response: {response_text}"
-                    _LOGGER.error(log)
-                    raise AirOSDeviceConnectionError from None
+                    _LOGGER.error(
+                        "Status API call failed with status %d: %s",
+                        response.status,
+                        response_text,
+                    )
+                    raise AirOSDeviceConnectionError
         except (
             aiohttp.ClientError,
             aiohttp.client_exceptions.ConnectionTimeoutError,
         ) as err:
-            _LOGGER.exception("Error during authenticated status.cgi call")
+            _LOGGER.error("Status API call failed: %s", err)
             raise AirOSDeviceConnectionError from err
 
     async def stakick(self, mac_address: str = None) -> bool:

airos/data.py

@@ -2,13 +2,41 @@
 
 from dataclasses import dataclass
 from enum import Enum
+import ipaddress
 import logging
+import re
 from typing import Any
 
 from mashumaro import DataClassDictMixin
 
 logger = logging.getLogger(__name__)
 
+# Regex for a standard MAC address format (e.g., 01:23:45:67:89:AB)
+# This handles both colon and hyphen separators.
+MAC_ADDRESS_REGEX = re.compile(r"^([0-9a-fA-F]{2}[:-]){5}([0-9a-fA-F]{2})$")
+
+# Regex for a MAC address mask (e.g., the redacted format 00:00:00:00:89:AB)
+MAC_ADDRESS_MASK_REGEX = re.compile(r"^(00:){4}[0-9a-fA-F]{2}[:-][0-9a-fA-F]{2}$")
+
+
+def is_mac_address(value: str) -> bool:
+    """Check if a string is a valid MAC address."""
+    return bool(MAC_ADDRESS_REGEX.match(value))
+
+
+def is_mac_address_mask(value: str) -> bool:
+    """Check if a string is a valid MAC address mask (e.g., the redacted format)."""
+    return bool(MAC_ADDRESS_MASK_REGEX.match(value))
+
+
+def is_ip_address(value: str) -> bool:
+    """Check if a string is a valid IPv4 or IPv6 address."""
+    try:
+        ipaddress.ip_address(value)
+        return True
+    except ValueError:
+        return False
+
 
 def _check_and_log_unknown_enum_value(
     data_dict: dict[str, Any],
@@ -31,6 +59,98 @@ def _check_and_log_unknown_enum_value(
             del data_dict[key]
 
 
+def redact_data_smart(data: dict) -> dict:
+    """Recursively redacts sensitive keys in a dictionary."""
+    sensitive_keys = {
+        "hostname",
+        "essid",
+        "mac",
+        "apmac",
+        "hwaddr",
+        "lastip",
+        "ipaddr",
+        "ip6addr",
+        "device_id",
+        "sys_id",
+        "station_id",
+        "platform",
+    }
+
+    def _redact(d: dict):
+        if not isinstance(d, dict):
+            return d
+
+        redacted_d = {}
+        for k, v in d.items():
+            if k in sensitive_keys:
+                if isinstance(v, str) and (is_mac_address(v) or is_mac_address_mask(v)):
+                    # Redact only the first 6 hex characters of a MAC address
+                    redacted_d[k] = "00:00:00:00:" + v.replace("-", ":").upper()[-5:]
+                elif isinstance(v, str) and is_ip_address(v):
+                    # Redact to a dummy local IP address
+                    redacted_d[k] = "127.0.0.3"
+                elif isinstance(v, list) and all(
+                    isinstance(i, str) and is_ip_address(i) for i in v
+                ):
+                    # Redact list of IPs to a dummy list
+                    redacted_d[k] = ["127.0.0.3"]
+                else:
+                    redacted_d[k] = "REDACTED"
+            elif isinstance(v, dict):
+                redacted_d[k] = _redact(v)
+            elif isinstance(v, list):
+                redacted_d[k] = [
+                    _redact(item) if isinstance(item, dict) else item for item in v
+                ]
+            else:
+                redacted_d[k] = v
+        return redacted_d
+
+    return _redact(data)
+
+
+def _redact_ip_addresses(addresses: str | list[str]) -> str | list[str]:
+    """Redacts the first three octets of an IPv4 address."""
+    if isinstance(addresses, str):
+        addresses = [addresses]
+
+    redacted_list = []
+    for ip in addresses:
+        try:
+            parts = ip.split(".")
+            if len(parts) == 4:
+                # Keep the last octet, but replace the rest with a placeholder.
+                redacted_list.append(f"127.0.0.{parts[3]}")
+            else:
+                # Handle non-standard IPs or IPv6 if it shows up here
+                redacted_list.append("REDACTED")
+        except (IndexError, ValueError):
+            # In case the IP string is malformed
+            redacted_list.append("REDACTED")
+
+    return redacted_list if isinstance(addresses, list) else redacted_list[0]
+
+
+def _redact_mac_addresses(macs: str | list[str]) -> str | list[str]:
+    """Redacts the first four octets of a MAC address."""
+    if isinstance(macs, str):
+        macs = [macs]
+
+    redacted_list = []
+    for mac in macs:
+        try:
+            parts = mac.split(":")
+            if len(parts) == 6:
+                # Keep the last two octets, replace the rest with a placeholder
+                redacted_list.append(f"00:11:22:33:{parts[4]}:{parts[5]}")
+            else:
+                redacted_list.append("REDACTED")
+        except (IndexError, ValueError):
+            redacted_list.append("REDACTED")
+
+    return redacted_list if isinstance(macs, list) else redacted_list[0]
+
+
 class IeeeMode(Enum):
     """Enum definition."""
 
@@ -152,6 +272,7 @@ class Polling:
     fixed_frame: bool
     gps_sync: bool
     ff_cap_rep: bool
+    flex_mode: int | None = None  # Not present in all devices
 
 
 @dataclass
@@ -207,9 +328,14 @@ class EthList:
 class GPSData:
     """Leaf definition."""
 
-    lat: str
-    lon: str
-    fix: int
+    lat: str | None = None
+    lon: str | None = None
+    fix: int | None = None
+    sats: int | None = None  # LiteAP GPS
+    dim: int | None = None  # LiteAP GPS
+    dop: float | None = None  # LiteAP GPS
+    alt: float | None = None  # LiteAP GPS
+    time_synced: int | None = None  # LiteAP GPS
 
 
 @dataclass
@@ -235,7 +361,6 @@ class Remote:
     totalram: int
     freeram: int
     netrole: str
-    mode: WirelessMode
     sys_id: str
     tx_throughput: int
     rx_throughput: int
@@ -254,20 +379,21 @@ class Remote:
     rx_bytes: int
     antenna_gain: int
     cable_loss: int
-    height: int
     ethlist: list[EthList]
     ipaddr: list[str]
-    ip6addr: list[str]
     gps: GPSData
     oob: bool
     unms: UnmsStatus
     airview: int
     service: ServiceTime
+    mode: WirelessMode | None = None  # Investigate why remotes can have no mode set
+    ip6addr: list[str] | None = None  # For v4 only devices
+    height: int | None = None
 
     @classmethod
     def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]:
         """Pre-deserialize hook for Wireless."""
-        _check_and_log_unknown_enum_value(d, "mode", WirelessMode, "Wireless", "mode")
+        _check_and_log_unknown_enum_value(d, "mode", WirelessMode, "Remote", "mode")
         return d
 
 
@@ -329,7 +455,6 @@ class Wireless:
     """Leaf definition."""
 
     essid: str
-    mode: WirelessMode
     ieeemode: IeeeMode
     band: int
     compat_11n: int
@@ -362,6 +487,7 @@ class Wireless:
     count: int
     sta: list[Station]
     sta_disconnected: list[Disconnected]
+    mode: WirelessMode | None = None  # Investigate further (see WirelessMode in Remote)
 
     @classmethod
     def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]:

airos/discovery.py

@@ -276,7 +276,9 @@ def parse_airos_packet(self, data: bytes, host_ip: str) -> dict[str, Any] | None
         return parsed_info
 
 
-async def async_discover_devices(timeout: int) -> dict[str, dict[str, Any]]:
+async def airos_discover_devices(
+    timeout: int = 30, listen_ip: str = "0.0.0.0", port: int = DISCOVERY_PORT
+) -> dict[str, dict[str, Any]]:
     """Discover unconfigured airOS devices on the network for a given timeout.
 
     This function sets up a listener, waits for a period, and returns
@@ -301,7 +303,7 @@ async def _async_airos_device_found(device_info: dict[str, Any]) -> None:
             protocol,
         ) = await asyncio.get_running_loop().create_datagram_endpoint(
             lambda: AirOSDiscoveryProtocol(_async_airos_device_found),
-            local_addr=("0.0.0.0", DISCOVERY_PORT),
+            local_addr=(listen_ip, port),
         )
         try:
             await asyncio.sleep(timeout)