hermes-agent/tests/agent/test_local_stream_timeout.py
Teknium 8c3935ebe8
fix: is_local_endpoint misses Docker/Podman DNS names (#7950)
* fix(tools): neutralize shell injection in _write_to_sandbox via path quoting

_write_to_sandbox interpolated storage_dir and remote_path directly into
a shell command passed to env.execute(). Paths containing shell
metacharacters (spaces, semicolons, $(), backticks) could trigger
arbitrary command execution inside the sandbox.

Fix: wrap both paths with shlex.quote(). Clean paths (alphanumeric +
slashes/hyphens/dots) are left unmodified by shlex.quote, so existing
behavior is unchanged. Paths with unsafe characters get single-quoted.

Tests added for spaces, $(command) substitution, and semicolon injection.

* fix: is_local_endpoint misses Docker/Podman DNS names

host.docker.internal, host.containers.internal, gateway.docker.internal,
and host.lima.internal are well-known DNS names that container runtimes
use to resolve the host machine. Users running Ollama on the host with
the agent in Docker/Podman hit the default 120s stream timeout instead
of the bumped 1800s because these hostnames weren't recognized as local.

Add _CONTAINER_LOCAL_SUFFIXES tuple and suffix check in
is_local_endpoint(). Tests cover all three runtime families plus a
negative case for domains that merely contain the suffix as a substring.
2026-04-11 14:46:18 -07:00

108 lines
4.5 KiB
Python

"""Tests for local provider stream read timeout auto-detection.
When a local LLM provider is detected (Ollama, llama.cpp, vLLM, etc.),
the httpx stream read timeout should be automatically increased from the
default 60s to HERMES_API_TIMEOUT (1800s) to avoid premature connection
kills during long prefill phases.
"""
import os
import pytest
from unittest.mock import patch
from agent.model_metadata import is_local_endpoint
class TestLocalStreamReadTimeout:
"""Verify stream read timeout auto-detection logic."""
@pytest.mark.parametrize("base_url", [
"http://localhost:11434",
"http://127.0.0.1:8080",
"http://0.0.0.0:5000",
"http://192.168.1.100:8000",
"http://10.0.0.5:1234",
"http://host.docker.internal:11434",
"http://host.containers.internal:11434",
"http://host.lima.internal:11434",
])
def test_local_endpoint_bumps_read_timeout(self, base_url):
"""Local endpoint + default timeout -> bumps to base_timeout."""
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("HERMES_STREAM_READ_TIMEOUT", None)
_base_timeout = float(os.getenv("HERMES_API_TIMEOUT", 1800.0))
_stream_read_timeout = float(os.getenv("HERMES_STREAM_READ_TIMEOUT", 120.0))
if _stream_read_timeout == 120.0 and base_url and is_local_endpoint(base_url):
_stream_read_timeout = _base_timeout
assert _stream_read_timeout == 1800.0
def test_user_override_respected_for_local(self):
"""User sets HERMES_STREAM_READ_TIMEOUT -> keep their value even for local."""
with patch.dict(os.environ, {"HERMES_STREAM_READ_TIMEOUT": "300"}, clear=False):
_base_timeout = float(os.getenv("HERMES_API_TIMEOUT", 1800.0))
_stream_read_timeout = float(os.getenv("HERMES_STREAM_READ_TIMEOUT", 120.0))
base_url = "http://localhost:11434"
if _stream_read_timeout == 120.0 and base_url and is_local_endpoint(base_url):
_stream_read_timeout = _base_timeout
assert _stream_read_timeout == 300.0
@pytest.mark.parametrize("base_url", [
"https://api.openai.com",
"https://openrouter.ai/api",
"https://api.anthropic.com",
])
def test_remote_endpoint_keeps_default(self, base_url):
"""Remote endpoint -> keep 120s default."""
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("HERMES_STREAM_READ_TIMEOUT", None)
_base_timeout = float(os.getenv("HERMES_API_TIMEOUT", 1800.0))
_stream_read_timeout = float(os.getenv("HERMES_STREAM_READ_TIMEOUT", 120.0))
if _stream_read_timeout == 120.0 and base_url and is_local_endpoint(base_url):
_stream_read_timeout = _base_timeout
assert _stream_read_timeout == 120.0
def test_empty_base_url_keeps_default(self):
"""No base_url set -> keep 120s default."""
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("HERMES_STREAM_READ_TIMEOUT", None)
_base_timeout = float(os.getenv("HERMES_API_TIMEOUT", 1800.0))
_stream_read_timeout = float(os.getenv("HERMES_STREAM_READ_TIMEOUT", 120.0))
base_url = ""
if _stream_read_timeout == 120.0 and base_url and is_local_endpoint(base_url):
_stream_read_timeout = _base_timeout
assert _stream_read_timeout == 120.0
class TestIsLocalEndpoint:
"""Direct unit tests for is_local_endpoint."""
@pytest.mark.parametrize("url", [
"http://localhost:11434",
"http://127.0.0.1:8080",
"http://0.0.0.0:5000",
"http://[::1]:11434",
"http://192.168.1.100:8000",
"http://10.0.0.5:1234",
"http://172.17.0.1:11434",
])
def test_classic_local_addresses(self, url):
assert is_local_endpoint(url) is True
@pytest.mark.parametrize("url", [
"http://host.docker.internal:11434",
"http://host.docker.internal:8080/v1",
"http://gateway.docker.internal:11434",
"http://host.containers.internal:11434",
"http://host.lima.internal:11434",
])
def test_container_dns_names(self, url):
assert is_local_endpoint(url) is True
@pytest.mark.parametrize("url", [
"https://api.openai.com",
"https://openrouter.ai/api",
"https://api.anthropic.com",
"https://evil.docker.internal.example.com",
])
def test_remote_endpoints(self, url):
assert is_local_endpoint(url) is False