mirror of
https://github.com/smkrv/ha-text-ai.git
synced 2026-07-21 22:54:00 +08:00
SMKRV
a8a91972ba
Phase A — Critical: - Remove dual timeout stacking in coordinator._send_to_api - Add Gemini-specific asyncio.timeout (sync SDK via to_thread) - Store full text in history for context; cap per-field at 32KB on disk - Fix instance lookup to match by normalized_name - Add Bearer/sk-/x-api-key credential sanitization patterns Phase B — Dead code removal: - Merge async_ask_question/async_process_question into single method - Remove dead is_anthropic flag from coordinator and __init__ - Remove unused DEFAULT_TIMEOUT and API_TIMEOUT constants - Remove redundant _create_history_dir calls - Consolidate async_check_api to use provider registry Phase C — Config flow correctness: - Truncate name before uniqueness check (prevent post-truncation collisions) - Add async_set_unique_id + _abort_if_unique_id_configured - Extract shared _build_parameter_schema for ConfigFlow/OptionsFlow dedup Phase D — UX improvements: - Optimize history write: serialize from memory, single file write - Show last 5 history entries in sensor attributes (was 1) - Return actual error type in ask_question service response - Add dedicated api_key_required error for provider/endpoint changes - Pass config_entry to DataUpdateCoordinator (HA 2024.8+) Phase E — Cleanup: - Extract _apply_structured_output for OpenAI/DeepSeek dedup - Reduce ABSOLUTE_MAX_HISTORY_SIZE to 200 with Final annotation - Remove dead translation keys (queued, invalid_characters) - Migrate to _attr_has_entity_name = True - Add from __future__ import annotations to all modules - Remove redundant api_status sensor attribute - Add missing translation keys (last_model, last_timestamp, etc.) Review agent fixes: - Add archive file cleanup (max 3 archives) to prevent disk exhaustion - Per-entry storage cap (32KB per field) for history on disk
95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
"""
|
|
Utility functions for HA Text AI integration.
|
|
|
|
@license: CC BY-NC-SA 4.0 International
|
|
@author: SMKRV
|
|
@github: https://github.com/smkrv/ha-text-ai
|
|
@source: https://github.com/smkrv/ha-text-ai
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import socket
|
|
from typing import Any
|
|
from urllib.parse import urlparse
|
|
|
|
from homeassistant.const import CONF_API_KEY
|
|
from homeassistant.core import HomeAssistant
|
|
|
|
|
|
def normalize_name(name: str) -> str:
|
|
"""Normalize name to conform to HA naming convention using underscores."""
|
|
normalized = ''.join(c if c.isalnum() or c == '_' else '_' for c in name)
|
|
normalized = '_'.join(filter(None, normalized.split('_')))
|
|
return normalized.lower()
|
|
|
|
|
|
def safe_log_data(
|
|
data: dict[str, Any],
|
|
sensitive_keys: tuple[str, ...] = (CONF_API_KEY,),
|
|
) -> dict[str, Any]:
|
|
"""Filter sensitive keys from data for safe logging."""
|
|
return {k: "***" if k in sensitive_keys else v for k, v in data.items()}
|
|
|
|
|
|
class _RestrictedIPError(ValueError):
|
|
"""Raised when an IP address is in a restricted range."""
|
|
|
|
|
|
def _check_ip_restricted(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
|
|
"""Check if an IP address is in a restricted range."""
|
|
return (
|
|
addr.is_private
|
|
or addr.is_reserved
|
|
or addr.is_loopback
|
|
or addr.is_link_local
|
|
or addr.is_multicast
|
|
or addr.is_unspecified
|
|
)
|
|
|
|
|
|
async def validate_endpoint(hass: HomeAssistant, endpoint: str) -> str:
|
|
"""Validate API endpoint URL for security.
|
|
|
|
Ensures HTTPS-only and blocks private/reserved IP ranges (SSRF protection).
|
|
Uses async DNS resolution to avoid blocking the event loop.
|
|
Returns the validated endpoint stripped of trailing slashes.
|
|
|
|
Raises:
|
|
ValueError: If the endpoint fails validation.
|
|
"""
|
|
parsed = urlparse(endpoint)
|
|
|
|
if parsed.scheme not in ("https",):
|
|
raise ValueError("Only HTTPS endpoints are allowed")
|
|
|
|
hostname = parsed.hostname
|
|
if not hostname:
|
|
raise ValueError("Invalid endpoint URL: no hostname")
|
|
|
|
# Block private/reserved IPs (direct IP or resolved hostname)
|
|
try:
|
|
addr = ipaddress.ip_address(hostname)
|
|
if _check_ip_restricted(addr):
|
|
raise _RestrictedIPError("Private/reserved IP addresses are not allowed")
|
|
except _RestrictedIPError:
|
|
raise
|
|
except ValueError:
|
|
# Not an IP literal — resolve hostname and check all resolved IPs
|
|
# to prevent DNS rebinding attacks
|
|
try:
|
|
addrinfos = await hass.async_add_executor_job(
|
|
socket.getaddrinfo, hostname, None
|
|
)
|
|
for family, _type, _proto, _canonname, sockaddr in addrinfos:
|
|
ip_str = sockaddr[0]
|
|
resolved_addr = ipaddress.ip_address(ip_str)
|
|
if _check_ip_restricted(resolved_addr):
|
|
raise ValueError(
|
|
"Hostname resolves to a restricted IP range"
|
|
)
|
|
except socket.gaierror as err:
|
|
raise ValueError(f"Cannot resolve hostname: {hostname}") from err
|
|
|
|
return endpoint.rstrip("/")
|