Home Assistant (maximeallanic)

Dependency "pillow@10.0.0" has known CVEs:

Pattern "\.(?:get|post|put|delete|patch|request)\s*\([^)]*(?:url|uri|target|endpoint|host)[^)]*(?:req|input|param|args|f['"]|\+)" matched in source_code: ".request( method, url, headers=self.headers, **kwargs" (at position 10166)

Dependency "aiohttp@3.8.0" has known CVEs:

Dependency "mcp@1.0.0" has known CVEs:

Pattern "(?:sendNotification|send_notification|notify|emit)\s*\((?!.*(?:queue|throttle|rateLimit|backpressure|bounded))" matched in source_code: "send_notification(" (at position 29384)

Pattern "(?:delay|sleep|timeout|wait)\s*[:=]\s*(?:[^;]*(?:charCodeAt|charAt|bit|byte|\&\s*1|>>|<<|\[i\]))" matched in source_code: "timeout=aiohttp.ClientTimeout(total=30) ) return self async def __aexit__(self, exc_type, exc_val, exc_tb): await self._ws_close() if self.session: await self.session.close() # --- WebSocket methods --- async def _ws_ensure_connected(self): """Ensure WebSocket is connected and authenticated""" if self._ws is not None and not self._ws.closed: return if not self.session: raise RuntimeError("Client not initialized. Use async context manager.") logger.info(f"Connecting WebSocket to {self._ws_url}") self._ws = await self.session.ws_connect(self._ws_url) self._ws_authenticated = False self._ws_id = 0 # Wait for auth_required message msg = await self._ws.receive_json() if msg.get("type") != "auth_required": raise RuntimeError(f"Expected auth_required, got: {msg.get('type')}") # Send auth await self._ws.send_json({ "type": "auth", "access_token": self.token, }) # Wait for auth response msg = await self._ws.receive_json() if msg.get("type") != "auth_ok": raise RuntimeError(f"WebSocket auth failed: {msg}") self._ws_authenticated = True logger.info("WebSocket authenticated successfully") async def _ws_send_command(self, command: Dict[str, Any]) -> Dict[str, Any]: """Send a WebSocket command and wait for result""" await self._ws_ensure_connected() self._ws_id += 1 command["id"] = self._ws_id await self._ws.send_json(command) # Read messages until we get the response for our id while True: msg = await self._ws.receive_json() if msg.get("id") == self._ws_id: if not msg.get("success", False): error = msg.get("error", {}) raise RuntimeError( f"WebSocket command failed: {error.get('code', 'unknown')} - {error.get('message', str(error))}" ) return msg.get("result") async def _ws_close(self): """Close WebSocket connection""" if self._ws is not None and not self._ws.closed: await self._ws.close() self._ws = None self._ws_authenticated = False async def ws_command(self, type: str, data: Dict[str, Any] = None) -> Any: """Send a generic WebSocket command.""" command = {"type": type} if data: command.update(data) return await self._ws_send_command(command) async def get_dashboards(self) -> Any: return await self.ws_command("lovelace/dashboards/list") async def get_dashboard_config(self, url_path: Optional[str] = None) -> Any: cmd = {"type": "lovelace/config"} if url_path: cmd["url_path"] = url_path return await self._ws_send_command(cmd) async def save_dashboard_config(self, config: Dict, url_path: Optional[str] = None, force: bool = False) -> Any: cmd = {"type": "lovelace/config/save", "config": config} if url_path: cmd["url_path"] = url_path if force: cmd["force"] = True return await self._ws_send_command(cmd) async def create_dashboard(self, title: str, url_path: str, icon: Optional[str] = None, show_in_sidebar: bool = True, require_admin: bool = False) -> Any: cmd = {"type": "lovelace/dashboards/create", "title": title, "url_path": url_path, "mode": "storage", "show_in_sidebar": show_in_sidebar, "require_admin": require_admin} if icon: cmd["icon"] = icon return await self._ws_send_command(cmd) async def delete_dashboard(self, dashboard_id: str) -> Any: return await self._ws_send_command({"type": "lovelace/dashboards/delete", "dashboard_id": dashboard_id}) async def get_lovelace_resources(self) -> Any: return await self.ws_command("lovelace/resources") # Built-in HA Lovelace cards (static list, current as of HA 2024+) BUILTIN_CARDS = [ "alarm-panel", "area", "button", "calendar", "camera", "conditional", "energy-carbon-consumed-gauge", "energy-date-selection", "energy-devices-detail-graph", "energy-devices-graph", "energy-distribution", "energy-gas-gauge", "energy-grid-neutrality-gauge", "energy-sources-table", "energy-solar-consumed-gauge", "energy-solar-graph", "energy-summary", "energy-usage-graph", "energy-water-gauge", "entities", "entity", "entity-filter", "gauge", "glance", "grid", "history-graph", "horizontal-stack", "humidifier", "iframe", "light", "logbook", "map", "markdown", "media-control", "picture", "picture-elements", "picture-entity", "picture-glance", "plant-status", "sensor", "shopping-list", "statistic", "statistics-graph", "thermostat", "tile", "todo-list", "vertical-stack", "weather-forecast", "webpage", ] async def list_lovelace_cards(self) -> Dict[str, Any]: """List all available Lovelace cards: built-in, registered resources, and HACS plugins.""" result: Dict[str, Any] = { "builtin_cards": [f"card-{c}" for c in self.BUILTIN_CARDS], "registered_resources": [], "hacs_plugins": [], } # Registered Lovelace resources try: resources = await self.get_lovelace_resources() result["registered_resources"] = resources or [] except Exception as e: result["registered_resources_error"] = str(e) # HACS installed plugins try: repos = await self.hacs_list_repositories(categories=["plugin"]) result["hacs_plugins"] = [ { "name": r.get("name"), "full_name": r.get("full_name"), "installed": r.get("installed", False), "version_installed": r.get("installed_version") or r.get("version_installed"), "version_available": r.get("available_version") or r.get("last_version"), "description": r.get("description"), } for r in (repos if isinstance(repos, list) else []) if r.get("installed") ] except Exception as e: result["hacs_plugins_error"] = str(e) return result # --- HACS methods --- async def hacs_status(self) -> Any: return await self.ws_command("hacs/info") async def hacs_list_repositories(self, categories: Optional[List[str]] = None) -> Any: data: Dict[str, Any] = {} if categories: data["categories"] = categories return await self.ws_command("hacs/repositories/list", data or None) async def hacs_repository_info(self, repository_id: str) -> Any: return await self.ws_command("hacs/repository/info", {"repository_id": repository_id}) async def hacs_download(self, repository_id: str, version: Optional[str] = None) -> Any: data: Dict[str, Any] = {"repository": repository_id} if version: data["version"] = version return await self.ws_command("hacs/repository/download", data) async def hacs_remove(self, repository_id: str) -> Any: return await self.ws_command("hacs/repository/remove", {"repository": repository_id}) async def hacs_refresh(self, repository_id: str) -> Any: return await self.ws_command("hacs/repository/refresh", {"repository": repository_id}) async def hacs_releases(self, repository_id: str) -> Any: return await self.ws_command("hacs/repository/releases", {"repository_id": repository_id}) async def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]: """Make HTTP request to Home Assistant API""" if not self.session: raise RuntimeError("Client not initialized. Use async context manager.") url = urljoin(self.base_url, endpoint) try: async with self.session.request( method, url, headers=self.headers, **kwargs ) as response: response.raise_for_status() return await response.json() except aiohttp.ClientError as e: logger.error(f"HTTP request failed: {e}") raise except Exception as e: logger.error(f"Request failed: {e}") raise async def get_states(self) -> List[Dict[str, Any]]: """Get all entity states""" return await self._request("GET", "/api/states") async def get_state(self, entity_id: str) -> Dict[str, Any]: """Get state of specific entity""" return await self._request("GET", f"/api/states/{entity_id}") async def call_service(self, domain: str, service: str, service_data: Optional[Dict] = None) -> List[Dict[str, Any]]: """Call a Home Assistant service""" endpoint = f"/api/services/{domain}/{service}" data = service_data or {} return await self._request("POST", endpoint, json=data) async def get_config(self) -> Dict[str, Any]: """Get Home Assistant configuration""" return await self._request("GET", "/api/config") async def get_services(self) -> List[Dict[str, Any]]: """Get all available services""" return await self._request("GET", "/api/services") async def get_events(self) -> List[Dict[str, Any]]: """Get event types""" return await self._request("GET", "/api/events") async def fire_event(self, event_type: str, event_data: Optional[Dict] = None) -> Dict[str, Any]: """Fire an event""" endpoint = f"/api/events/{event_type}" data = event_data or {} return await self._request("POST", endpoint, json=data) async def get_history(self, start_time: Optional[str] = None, end_time: Optional[str] = None, filter_entity_id: Optional[str] = None, minimal_response: bool = False, no_attributes: bool = False, significant_changes_only: bool = False) -> List[Dict[str, Any]]: """Get historical data""" if start_time: endpoint = f"/api/history/period/{start_time}" else: endpoint = "/api/history/period" params = {} if end_time: params["end_time"] = end_time if filter_entity_id: params["filter_entity_id"] = filter_entity_id if minimal_response: params["minimal_response"] = "true" if no_attributes: params["no_attributes"] = "true" if significant_changes_only: params["significant_changes_only"] = "true" return await self._request("GET", endpoint, params=params) async def get_logbook(self, start_time: Optional[str] = None, end_time: Optional[str] = None, entity: Optional[str] = None) -> List[Dict[str, Any]]: """Get logbook entries""" if start_time: endpoint = f"/api/logbook/{start_time}" else: endpoint = "/api/logbook" params = {} if end_time: params["end_time"] = end_time if entity: params["entity"] = entity return await self._request("GET", endpoint, params=params) async def get_error_log(self) -> str: """Get error log""" response = await self._request("GET", "/api/error_log") return response if isinstance(response, str) else str(response) async def get_calendars(self) -> List[Dict[str, Any]]: """Get list of calendar entities""" return await self._request("GET", "/api/calendars") async def get_calendar_events(self, calendar_id: str, start: str, end: str) -> List[Dict[str, Any]]: """Get calendar events for a specific calendar""" endpoint = f"/api/calendars/{calendar_id}" params = {"start": start, "end": end} return await self._request("GET", endpoint, params=params) async def get_camera_proxy(self, camera_entity_id: str) -> byte" (at position 1920)