Added input validation and fixed linting

glocaltokens/client.py

@@ -17,6 +17,7 @@
     ACCESS_TOKEN_DURATION,
     ACCESS_TOKEN_SERVICE,
     ANDROID_ID_LENGTH,
+    DEFAULT_DISCOVERY_PORT,
     DISCOVERY_TIMEOUT,
     GOOGLE_HOME_FOYER_API,
     HOMEGRAPH_DURATION,
@@ -27,6 +28,7 @@
 from .types import DeviceDict
 from .utils import network as net_utils, token as token_utils
 from .utils.logs import censor
+from .utils.network import is_valid_ipv4_address
 
 logging.basicConfig(level=logging.ERROR)
 LOGGER = logging.getLogger(__name__)
@@ -367,6 +369,24 @@ def get_google_devices(
 
         devices: list[Device] = []
 
+        def is_dict_with_valid_ipv4_addresses(data: Dict[str, str]) -> bool:
+            # Test if the data structure is correct and if each entry contains a
+            # valid IPv4 address.
+            return isinstance(data, dict) and all(
+                isinstance(x, str) and is_valid_ipv4_address(x) for x in data.values()
+            )
+
+        if discovery_address_dict and not is_dict_with_valid_ipv4_addresses(
+            discovery_address_dict
+        ):
+            # We need to disable flake8-use-fstring, because of the brackets,
+            # it causes a false positive.
+            LOGGER.error(
+                "Invalid dictionary structure for discovery_address_dict "
+                "argument. Correct structure is {'device_name': 'ipaddress'}"  # noqa
+            )
+            return devices
+
         if homegraph is None:
             LOGGER.debug("Failed to fetch homegraph")
             return devices
@@ -387,9 +407,7 @@ def find_device(unique_id: str) -> NetworkDevice | None:
                     return device
             return None
 
-        discovery_address_dict = (
-            discovery_address_dict if discovery_address_dict else {}
-        )
+        address_dict = discovery_address_dict if discovery_address_dict else {}
 
         LOGGER.debug("Iterating in %d homegraph devices", len(homegraph.home.devices))
         for item in homegraph.home.devices:
@@ -411,12 +429,11 @@ def find_device(unique_id: str) -> NetworkDevice | None:
                         unique_id,
                     )
                     network_device = find_device(unique_id)
-                elif item.device_name in discovery_address_dict:
+                elif item.device_name in address_dict:
                     network_device = NetworkDevice(
                         name=item.device_name,
-                        ip_address=discovery_address_dict[item.device_name]
-                        or "0.0.0.0",
-                        port=0,
+                        ip_address=address_dict[item.device_name],
+                        port=DEFAULT_DISCOVERY_PORT,
                         model=item.hardware.model,
                         unique_id=item.device_info.device_id,
                     )
@@ -452,7 +469,7 @@ def get_google_devices_json(
         self,
         models_list: list[str] | None = None,
         indent: int = 2,
-        address_list: Dict[str, str] | None = None,
+        discovery_address_dict: Dict[str, str] | None = None,
         zeroconf_instance: Zeroconf | None = None,
         force_homegraph_reload: bool = False,
     ) -> str:
@@ -462,16 +479,17 @@ def get_google_devices_json(
 
         models_list: The list of accepted model names.
         indent: The indentation for the json formatting.
-        address_list: Whether or not the device's IP and port should
-          be searched for in the network.
+        discovery_address_dict: Dict of network devices from the local network
+          ({"name": "ip_address"}). If set to `None` will try to automatically
+          discover network devices. Disable discovery by setting to `{}`.
         zeroconf_instance: If you already have an initialized zeroconf instance,
           use it here.
         force_homegraph_reload: If the stored homegraph should be generated again.
         """
 
         google_devices = self.get_google_devices(
             models_list=models_list,
-            discovery_address_dict=address_list,
+            discovery_address_dict=discovery_address_dict,
             zeroconf_instance=zeroconf_instance,
             force_homegraph_reload=force_homegraph_reload,
         )

glocaltokens/const.py

@@ -18,6 +18,7 @@
 HOMEGRAPH_DURATION: Final = 24 * 60 * 60
 
 DISCOVERY_TIMEOUT: Final = 2
+DEFAULT_DISCOVERY_PORT: Final = 0
 
 GOOGLE_HOME_MODELS: Final = [
     "Google Home",

tests/test_client.py

@@ -386,7 +386,7 @@ def test_get_google_devices_json(
         )
         m_get_google_devices.return_value = [google_device]
 
-        json_string = self.client.get_google_devices_json(address_list={})
+        json_string = self.client.get_google_devices_json(discovery_address_dict={})
         self.assertEqual(m_get_google_devices.call_count, 1)
         received_json = json.loads(json_string)
         received_device = received_json[0]