fix: Round 2 review findings — async SSRF, error sanitization, constants

Security:
- Make validate_endpoint async with hass.async_add_executor_job for DNS resolution
- Use _RestrictedIPError instead of fragile string matching for IP check flow
- Add is_multicast/is_unspecified to SSRF IP restriction checks
- Remove resolved private IP from error messages (generic message)
- Remove raw endpoint from __init__.py error log
- Sanitize error messages in metrics: strip URLs, API keys, Gemini key patterns
- Truncate API error response bodies before logging (512 chars)
- Use generic error messages in Gemini exception handlers (no str(e) interpolation)
- Pass api_key as explicit APIClient constructor parameter (not from header)
- Add defensive validation: Gemini provider requires api_key at construction

Code quality:
- Add constants: DEFAULT_INSTANCE_NAME, MIN/MAX_CONTEXT_MESSAGES, MIN/MAX_HISTORY_SIZE
- Replace all hardcoded schema ranges with named constants
- Move datetime import to module level in history.py
- Remove unused full_history/full_history_available keys
- Chain socket.gaierror properly with raise...from
This commit is contained in:
SMKRV
2026-03-12 01:57:49 +03:00
parent c54bfcff3b
commit 31560a8835
8 changed files with 72 additions and 38 deletions
+30 -16
View File
@@ -12,6 +12,7 @@ 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:
@@ -29,11 +30,27 @@ def safe_log_data(
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 validate_endpoint(endpoint: str) -> str:
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:
@@ -51,28 +68,25 @@ def validate_endpoint(endpoint: str) -> str:
# Block private/reserved IPs (direct IP or resolved hostname)
try:
addr = ipaddress.ip_address(hostname)
if addr.is_private or addr.is_reserved or addr.is_loopback or addr.is_link_local:
raise ValueError("Private/reserved IP addresses are not allowed")
except ValueError as e:
if "not allowed" in str(e):
raise
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 = socket.getaddrinfo(hostname, None)
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 (
resolved_addr.is_private
or resolved_addr.is_reserved
or resolved_addr.is_loopback
or resolved_addr.is_link_local
):
if _check_ip_restricted(resolved_addr):
raise ValueError(
f"Hostname {hostname} resolves to private/reserved IP {ip_str}"
"Hostname resolves to a restricted IP range"
)
except socket.gaierror:
raise ValueError(f"Cannot resolve hostname: {hostname}")
except socket.gaierror as err:
raise ValueError(f"Cannot resolve hostname: {hostname}") from err
return endpoint.rstrip("/")