mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
f2a8887815
Add process-local SSH forwarding for custom OpenAI-compatible endpoints. Persist optional SSH settings through the endpoint API and Desktop form, then rewrite the endpoint only at runtime for CLI, TUI, and Desktop.
73 lines
2.1 KiB
Python
73 lines
2.1 KiB
Python
from hermes_cli import ssh_tunnel
|
|
|
|
|
|
def test_tunnel_config_accepts_user_at_host_and_explicit_key():
|
|
config = ssh_tunnel.SshTunnelConfig.from_dict({
|
|
"host": "ari@example.test",
|
|
"port": 2222,
|
|
"key_path": "/tmp/key",
|
|
})
|
|
|
|
assert config == ssh_tunnel.SshTunnelConfig(
|
|
host="example.test", user="ari", port=2222, key_path="/tmp/key"
|
|
)
|
|
agent_config = ssh_tunnel.SshTunnelConfig.from_dict({"host": "example.test"})
|
|
assert agent_config is not None
|
|
assert agent_config.key_path == ""
|
|
|
|
|
|
def test_tunnel_url_rewrites_only_the_runtime_host(monkeypatch):
|
|
ssh_tunnel.close_ssh_tunnels()
|
|
started = []
|
|
|
|
class LiveProcess:
|
|
def __init__(self):
|
|
self.alive = True
|
|
|
|
def poll(self):
|
|
return None if self.alive else 0
|
|
|
|
def terminate(self):
|
|
self.alive = False
|
|
|
|
def wait(self, timeout=None):
|
|
return 0
|
|
|
|
def kill(self):
|
|
self.alive = False
|
|
|
|
def fake_start(self):
|
|
if self.local_port and self.process and self.process.poll() is None:
|
|
return self.local_port
|
|
started.append(self)
|
|
self.local_port = 45678
|
|
self.process = LiveProcess()
|
|
return self.local_port
|
|
|
|
monkeypatch.setattr(ssh_tunnel.ManagedSshTunnel, "start", fake_start)
|
|
|
|
url = ssh_tunnel.resolve_ssh_tunnel_url(
|
|
"http://127.0.0.1:30090/v1",
|
|
{"host": "remote.example", "user": "ari", "key_path": "/tmp/key"},
|
|
)
|
|
|
|
assert url == "http://127.0.0.1:45678/v1"
|
|
assert len(started) == 1
|
|
assert started[0].remote_host == "127.0.0.1"
|
|
assert started[0].remote_port == 30090
|
|
|
|
assert ssh_tunnel.resolve_ssh_tunnel_url(
|
|
"http://127.0.0.1:30090/v1",
|
|
{"host": "remote.example", "user": "ari", "key_path": "/tmp/key"},
|
|
) == url
|
|
assert len(started) == 1
|
|
ssh_tunnel.close_ssh_tunnels()
|
|
|
|
|
|
def test_tunnel_config_rejects_user_twice():
|
|
try:
|
|
ssh_tunnel.SshTunnelConfig.from_dict({"host": "ari@example.test", "user": "other"})
|
|
except ValueError as exc:
|
|
assert "must not include a user" in str(exc)
|
|
else:
|
|
raise AssertionError("unsafe SSH target accepted")
|