Merge pull request #2340 from AppDaemon/websocket-hardening

hass plugin match/case structures

Author: acockburn

appdaemon/models/config/plugin.py

@@ -92,7 +92,7 @@ class HASSConfig(PluginConfig):
     enable_started_event: bool = True
     """If true, the plugin will wait for the 'homeassistant_started' event before starting the plugin."""
     cert_path: CoercedPath | None = None
-    cert_verify: bool | None = None
+    cert_verify: bool = True
     commtype: str = "WS"
     q_timeout: int = 30
     ws_timeout: Annotated[

appdaemon/plugins/hass/hassapi.py

@@ -58,10 +58,14 @@ def __init__(self, ad: AppDaemon, config_model: "AppConfig"):
         self.register_constraint("constrain_input_select")
 
     @utils.sync_decorator
-    async def ping(self) -> float:
+    async def ping(self) -> float | None:
         """Gets the number of seconds """
         if (plugin := self._plugin) is not None:
-            return (await plugin.ping())['ad_duration']
+            match await plugin.ping():
+                case {"ad_status": "OK", "ad_duration": ad_duration}:
+                    return ad_duration
+                case _:
+                    return None
 
     @utils.sync_decorator
     async def check_for_entity(self, entity_id: str, namespace: str | None = None) -> bool:
@@ -539,16 +543,14 @@ def get_service_info(self, service: str) -> dict | None:
         Returns:
             Information about the service in a dict with the following keys: ``name``, ``description``, ``target``, and
             ``fields``.
-
         """
-        if (plugin := self._plugin) is not None:
-            domain, service_name = service.split("/", 2)
-            for service_def in plugin.services:
-                if service_def.get("domain") == domain:
-                    if (services := service_def.get("services")) is not None:
-                        return deepcopy(services.get(service_name))
-            else:
-                self.logger.warning("Service info not found for domain '%s", domain)
+        match self._plugin:
+            case HassPlugin() as plugin:
+                domain, service_name = service.split("/", 2)
+                if info := plugin.services.get(domain, {}).get(service_name):
+                    # Return a copy of the info dict to prevent accidental modification
+                    return deepcopy(info)
+        self.logger.warning("Service info not found for domain '%s", domain)
 
     # Methods that use self.call_service
 
@@ -680,7 +682,7 @@ async def get_history(
         significant_changes_only: bool | None = None,
         callback: Callable | None = None,
         namespace: str | None = None,
-    ) -> list[list[dict[str, Any]]]:
+    ) -> list[list[dict[str, Any]]] | None:
         """Gets access to the HA Database.
         This is a convenience function that allows accessing the HA Database, so the
         history state of a device can be retrieved. It allows for a level of flexibility
@@ -755,31 +757,27 @@ async def get_history(
             end_time = end_time or await self.get_now()
             start_time = end_time - timedelta(days=days)
 
-
-        plugin: "HassPlugin" = self.AD.plugins.get_plugin_object(
-            namespace or self.namespace
-        )
-
-        if plugin is not None:
-            coro = plugin.get_history(
-                filter_entity_id=entity_id,
-                timestamp=start_time,
-                end_time=end_time,
-                minimal_response=minimal_response,
-                no_attributes=no_attributes,
-                significant_changes_only=significant_changes_only,
-            )
-
-            if callback is not None and callable(callback):
-                self.create_task(coro, callback)
-            else:
-                return await coro
-
-        else:
-            self.logger.warning(
-                "Wrong Namespace selected, as %s has no database plugin attached to it",
-                namespace,
-            )
+        plugin = self.AD.plugins.get_plugin_object(namespace or self.namespace)
+        match plugin:
+            case HassPlugin():
+                coro = plugin.get_history(
+                    filter_entity_id=entity_id,
+                    timestamp=start_time,
+                    end_time=end_time,
+                    minimal_response=minimal_response,
+                    no_attributes=no_attributes,
+                    significant_changes_only=significant_changes_only,
+                )
+
+                if callback is not None and callable(callback):
+                    self.create_task(coro, callback)
+                else:
+                    return await coro
+            case _:
+                self.logger.warning(
+                    "Wrong Namespace selected, as %s has no database plugin attached to it",
+                    namespace,
+                )
 
     @utils.sync_decorator
     async def get_logbook(