mirror of
https://github.com/smkrv/ha-text-ai.git
synced 2026-07-22 07:03:58 +08:00
feat: Add allow_local_network option for self-hosted LLM proxies
Add per-instance boolean option to allow private IP endpoints and HTTP scheme for self-hosted LLM proxies (LiteLLM, Ollama, vLLM, etc.). - New CONF_ALLOW_LOCAL_NETWORK config option (default: false) - When enabled: allows RFC1918 private IPs and HTTP endpoints - When disabled: full SSRF protection preserved (HTTPS + public IPs only) - Multicast and unspecified addresses blocked regardless of setting - Warning logged when local network mode is active - Checkbox added to ConfigFlow and OptionsFlow UI - Translations for all 8 languages Closes #9
This commit is contained in:
@@ -48,10 +48,11 @@ def _check_ip_restricted(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) ->
|
||||
)
|
||||
|
||||
|
||||
async def validate_endpoint(hass: HomeAssistant, endpoint: str) -> str:
|
||||
async def validate_endpoint(hass: HomeAssistant, endpoint: str, *, allow_local: bool = False) -> str:
|
||||
"""Validate API endpoint URL for security.
|
||||
|
||||
Ensures HTTPS-only and blocks private/reserved IP ranges (SSRF protection).
|
||||
When allow_local is True, permits private IPs and HTTP scheme for self-hosted proxies.
|
||||
Uses async DNS resolution to avoid blocking the event loop.
|
||||
Returns the validated endpoint stripped of trailing slashes.
|
||||
|
||||
@@ -60,35 +61,65 @@ async def validate_endpoint(hass: HomeAssistant, endpoint: str) -> str:
|
||||
"""
|
||||
parsed = urlparse(endpoint)
|
||||
|
||||
if parsed.scheme not in ("https",):
|
||||
raise ValueError("Only HTTPS endpoints are allowed")
|
||||
if allow_local:
|
||||
if parsed.scheme not in ("https", "http"):
|
||||
raise ValueError("Only HTTPS and HTTP endpoints are allowed")
|
||||
else:
|
||||
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
|
||||
if allow_local:
|
||||
# Even in local mode, block multicast and unspecified addresses
|
||||
_is_ip_literal = True
|
||||
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
|
||||
addr = ipaddress.ip_address(hostname)
|
||||
except ValueError:
|
||||
_is_ip_literal = False
|
||||
|
||||
if _is_ip_literal:
|
||||
if addr.is_multicast or addr.is_unspecified:
|
||||
raise ValueError("Multicast and unspecified addresses are not allowed")
|
||||
else:
|
||||
# Not an IP literal — resolve and check
|
||||
try:
|
||||
addrinfos = await hass.async_add_executor_job(
|
||||
socket.getaddrinfo, hostname, None
|
||||
)
|
||||
for family, _type, _proto, _canonname, sockaddr in addrinfos:
|
||||
resolved = ipaddress.ip_address(sockaddr[0])
|
||||
if resolved.is_multicast or resolved.is_unspecified:
|
||||
raise ValueError(
|
||||
"Hostname resolves to multicast/unspecified address"
|
||||
)
|
||||
except socket.gaierror as err:
|
||||
raise ValueError(f"Cannot resolve hostname: {hostname}") from err
|
||||
else:
|
||||
# Full SSRF protection — block private/reserved IPs
|
||||
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("/")
|
||||
|
||||
Reference in New Issue
Block a user