Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion src/beaconmcp/proxmox/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -181,8 +181,11 @@ def api_call(self, node_name: str, method: str, path: str, **kwargs: Any) -> Any
# already running, and a second POST would start it again.
if method != "get" and not _never_sent(e):
self._cache().pop(node_name, None)
what = "did not answer in time" if isinstance(e, Timeout) else (
"dropped the connection mid-request"
)
return {
"error": f"Node '{node_name}' did not answer in time: {e}. "
"error": f"Node '{node_name}' {what}: {e}. "
"The request may still have been applied; check "
"proxmox_get_tasks before retrying."
}
Expand Down
28 changes: 21 additions & 7 deletions src/beaconmcp/proxmox/system.py
Original file line number Diff line number Diff line change
Expand Up @@ -187,10 +187,13 @@ def _agent_read_bytes(
count = _FILE_READ_MAX_BYTES + 1
result = client.get(node, api, file=path, count=count, decode=0)
legacy = False
if isinstance(result, dict) and "error" in result and "decode" in str(result["error"]):
if isinstance(result, dict) and "error" in result and any(
name in str(result["error"]) for name in ("decode", "count")
):
# PVE before the `decode` parameter: content arrives decoded, one
# JSON code point per byte.
result = client.get(node, api, file=path, count=count)
# JSON code point per byte. `count` may be just as new, so it is left
# out too; the caller's length check still enforces the limit.
result = client.get(node, api, file=path)
legacy = True
if isinstance(result, dict) and "error" in result:
return {"status": "error", "error": result["error"]}
Expand Down Expand Up @@ -321,10 +324,18 @@ async def connect_tcp(
f"Refusing to fetch from {host} ({local}): loopback and link-local "
"addresses are not allowed."
)
return await self._inner.connect_tcp(
addrs[0], port, timeout=timeout,
local_address=local_address, socket_options=socket_options,
)
# Every address was checked above, so trying them in turn keeps the
# guarantee: an unreachable AAAA on an IPv4-only host must not end it.
last: Exception | None = None
for addr in addrs:
try:
return await self._inner.connect_tcp(
addr, port, timeout=timeout,
local_address=local_address, socket_options=socket_options,
)
except (httpcore.ConnectError, httpcore.ConnectTimeout, OSError) as e:
last = e
raise httpcore.ConnectError(f"cannot connect to {host}: {last}") from last

async def sleep(self, seconds: float) -> None:
await self._inner.sleep(seconds)
Expand All @@ -337,6 +348,9 @@ def _fetch_transport() -> httpx.AsyncBaseTransport:
a proxy would resolve the name itself, out of the guard's reach.
"""
transport = httpx.AsyncHTTPTransport()
# httpx has no public hook for the network backend; `_pool` is the pool
# every request goes through (checked against httpx 0.28.1 / httpcore
# 1.0.9, and pinned by test_upload_links' rebinding test).
transport._pool = httpcore.AsyncConnectionPool(
ssl_context=httpx.create_ssl_context(),
network_backend=_GuardedNetworkBackend(),
Expand Down
7 changes: 5 additions & 2 deletions tests/test_agent_file_io.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,8 +41,11 @@ def get(self, node: str, path: str, **params: Any) -> Any:
if path == f"nodes/{node}/qemu/{VM}/status/current":
return {"status": "running"}
if path == f"nodes/{node}/qemu/{VM}/agent/file-read":
if "decode" in params and not self.has_decode_param:
return {"error": "400 Parameter verification failed: decode: property is not defined in schema"}
if not self.has_decode_param:
# Older qemu-server: neither `decode` nor `count` in the schema.
for name in ("decode", "count"):
if name in params:
return {"error": f"400 Parameter verification failed: {name}: property is not defined in schema"}
data = self.files[params["file"]][: params.get("count", 16 * 1024 * 1024)]
if params.get("decode", 1):
# Perl hands JSON the decoded bytes, one code point each.
Expand Down
7 changes: 7 additions & 0 deletions tests/test_proxmox_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -225,3 +225,10 @@ def test_get_is_retried_after_a_read_timeout(monkeypatch: pytest.MonkeyPatch) ->
ep = _Endpoint(ReadTimeout("read timed out"))
_client_with(monkeypatch, ep).get("pve1", "nodes")
assert ep.calls == 2


def test_a_dropped_write_is_not_called_a_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
ep = _Endpoint(RequestsConnectionError("Connection aborted: RemoteDisconnected"))
result = _client_with(monkeypatch, ep).post("pve1", "nodes")
assert ep.calls == 1
assert "dropped the connection mid-request" in result["error"]
23 changes: 23 additions & 0 deletions tests/test_upload_links.py
Original file line number Diff line number Diff line change
Expand Up @@ -544,3 +544,26 @@ async def test_guarded_connect_keeps_the_hostname(
)
assert res["status"] == "success", res
assert env.ssh.guest_files["/f"] == f"LOOPBACK-ONLY-SECRET files.test:{loopback_server}".encode()


async def test_fetch_tries_each_checked_address(monkeypatch) -> None:
# An unreachable first answer (an AAAA on an IPv4-only box) must not end
# the fetch: every address was checked, so the next one is still safe.
import httpcore

async def fake_dns(host):
return ["2001:db8::1", "198.51.100.7"]

monkeypatch.setattr(system, "_host_addresses", fake_dns)
backend = system._GuardedNetworkBackend()
tried: list[str] = []

async def connect(addr, port, **kw):
tried.append(addr)
if addr == "2001:db8::1":
raise httpcore.ConnectError("network unreachable")
return "stream"

monkeypatch.setattr(backend._inner, "connect_tcp", connect)
assert await backend.connect_tcp("example.com", 443) == "stream"
assert tried == ["2001:db8::1", "198.51.100.7"]
Loading