From d4f88afc9e223d5ada15b0145cac6882f041893c Mon Sep 17 00:00:00 2001 From: Pedro Henrique Penna Date: Mon, 12 Oct 2026 17:21:48 -0300 Subject: [PATCH] sandbox: bound managed control writes by the operation deadline Fixes #464. The managed control client sent every record without a deadline. exec, ping, and stop took their deadline only once they had sent their request, as the attach exchange did once it had sent the capability, and the Windows transport wrote each record with a synchronous os.write() to the named pipe. A control endpoint that authenticated the session and then stopped reading, such as a stalled OpenVMM, therefore held a request larger than the pipe's free inbound quota, and with it `sandbox exec`, `start`, or `stop`, past the --timeout that bounds the control operation, without limit. Only the closure of the pipe released it, and then with a broken pipe rather than a timeout. Reproduction ------------ The issue's self-contained reproduction authenticates the unmodified client over a real byte-mode named pipe with a 1 KiB quota, stops reading, and sends a valid 4,093-byte exec request with a 0.05 s response timeout. Against the previous control_session.py on Windows, with Python 3.14.7, the call was still blocked after 0.36 s and returned only when the reproduction closed the server, with BrokenPipeError. With this change, it fails with "managed control request timed out" after 0.06 s while the server keeps the pipe open. Fix --- - Each operation, the attach exchange, ping, exec, and stop, takes its deadline before it sends its request and bounds both the request and the response with it, and each transport's write_all() takes that deadline. A write that outlives it fails with the TimeoutError "managed control request timed out", as a read that outlives it already failed with "managed control response timed out". - The Windows transport opens the pipe for overlapped I/O and reads and writes it with overlapped transfers. A transfer that is still pending at its deadline is cancelled, and every transfer is waited for until it ends, since Windows uses its buffer until then, even when an exception such as KeyboardInterrupt ends the wait. The wait uses WaitForMultipleObjects(), which Ctrl+C interrupts on the main thread. Cancellation calls CancelIoEx() rather than Overlapped.cancel(), which marks a transfer as ended at once, so that a transfer that an exception releases at any point is waited for or keeps its buffer. Overlapped reads replace the PeekNamedPipe() polling, which bounded reads but has no counterpart for writes, and a read now completes as soon as data arrives rather than at the next 10 ms poll. ERROR_BROKEN_PIPE, ERROR_NO_DATA, and ERROR_PIPE_NOT_CONNECTED still report a closed endpoint. Connection retries keep their errno policy: Python maps the errors of CreateFile(), including the ERROR_PIPE_BUSY of the reconnect transition, to the errno values that the CRT gave for os.open(). - The socket transport sends in deadline-bounded slices rather than with one sendall(), which the socket timeout that the last read had left, at most 0.25 s, bounded instead of the operation deadline. - On both transports, a write that moves no bytes reports a closed endpoint rather than retrying until the deadline, as an empty read does and as the previous named-pipe writer did. - A session closes when a record transfer fails in either direction, since the record may be cut short and any later record would then not frame correctly, and it then refuses further operations with "managed control session is closed". A refusal or reset that arrives as a complete record leaves the session usable, as the managed-lifecycle scenario requires, and close() is idempotent. - doc/usage.md now describes --timeout as bounding each managed control request and its response. Tests ----- - Over a real named pipe whose endpoint authenticates the session and then stops reading, a 4,093-byte exec request with a 0.25 s timeout fails with the request timeout within 2 s of the deadline, while the endpoint keeps the pipe open. The session then refuses a ping, and the endpoint receives less than the request before the pipe closes. Against the previous control_session.py, the request stayed blocked until the test endpoint drained the pipe after 10 s, and the operation then failed with a response timeout. - On Linux, a 60 KiB exec request to a Unix-domain socket peer that does not read fails the same way, and the peer receives part of the request and then the end of the stream. - With a fixed clock, attach, ping, exec, and stop each write with the deadline with which they then read, and a write or read that fails closes the session once and refuses later operations without another transfer. - An interrupted named-pipe write is cancelled before the KeyboardInterrupt propagates, and leaves nothing in the pipe. - On both transports, a write that moves no bytes reports a closed endpoint at once. - Named-pipe reads time out without losing data that arrives later, and report a closed endpoint only after the data that the server wrote before it closed the pipe. The connection retry test now stubs CreateFile(). Validation ---------- - On a Windows 11 workstation with Python 3.14.7, ruff, pyright for Linux and Windows, compileall, the validate-nvx unit suites, the adversarial tests, and the CLI help checks pass, and the control-session tests pass in 25 of 25 repeated runs. - Bare-metal Linux/KVM, Linux/MSHV on Azure Linux 3.0, and Windows/WHP on Windows Server 2025 hosts ran an isolated clone of this change with the release package that CI built for 8c3cc0abf, whose guest, kernel, and OpenVMM inputs equal this commit's, and with that CI run's Ubuntu scratch template. On each, the validate-nvx unit suites pass, the socket tests on Linux and the named-pipe tests on Windows included, and `test-microvm --scenario managed-lifecycle --scenario managed-exec-config --scenario sandbox-lifecycle` passes against real OpenVMM, on WHP in 4 of 4 runs. The control-session tests pass in 25 of 25 repeated runs on the KVM and WHP hosts. - On the WHP host, the issue's reproduction against dev's control_session.py was still blocked after 0.36 s and ended only when the server closed the pipe; against this change, it timed out after 0.05 s. The new stalled-endpoint test failed against dev's control_session.py after 10.3 s. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- doc/usage.md | 4 +- scripts/nvx_tools/control_session.py | 269 ++++++++++++------ scripts/test_control_session.py | 397 ++++++++++++++++++++++++++- scripts/test_microvm_tests.py | 9 +- 4 files changed, 586 insertions(+), 93 deletions(-) diff --git a/doc/usage.md b/doc/usage.md index fbb539e19..9bbdf0c3d 100644 --- a/doc/usage.md +++ b/doc/usage.md @@ -605,8 +605,8 @@ launches. | `--memory-max BYTES` | none | Set the workload cgroup memory limit. | | `--pids-max COUNT` | none | Set the workload cgroup process limit. | | `--memory-mib MIB` | `256` | Set guest memory in MiB. | -| `--timeout SECONDS` | `60` | Set the control response timeout for managed `start`, `exec`, and `stop`, and the longest time that `provision`, `start`, `stop`, and `deprovision` wait for another lifecycle transition of the sandbox. It must be finite and greater than 0. | -| `--exec-timeout-ms MILLISECONDS` | `0` | Set the managed `exec` guest workload timeout in the unsigned 32-bit range `0..4294967295`; zero disables the workload deadline. This is separate from the finite host `--timeout` response deadline. | +| `--timeout SECONDS` | `60` | Set the timeout for each managed control request and its response in `start`, `exec`, and `stop`, and the longest time that `provision`, `start`, `stop`, and `deprovision` wait for another lifecycle transition of the sandbox. It must be finite and greater than 0. | +| `--exec-timeout-ms MILLISECONDS` | `0` | Set the managed `exec` guest workload timeout in the unsigned 32-bit range `0..4294967295`; zero disables the workload deadline. This is separate from the finite host `--timeout` control deadline. | | `--cwd GUEST_PATH` | `/` | Set an absolute working directory inside the workload root for managed `exec`. It is resolved after the workload identity and root are applied; missing, inaccessible, or non-directory paths fail the workload launch. Default and layered environments point `PWD` at this directory; an exact replacement environment controls `PWD` itself. | | `--environment KEY=VALUE` | omitted | Set the exact managed `exec` environment. Repeat for multiple entries. Empty values, spaces, additional equals signs, and UTF-8 are preserved. Inline values are visible in the invoking host process arguments; use `--environment-file` for sensitive values. | | `--environment-file PATH` | omitted | Read the exact managed `exec` environment from a UTF-8 JSON array of `KEY=VALUE` strings, limited to 1 MiB of input. An empty array requests an empty environment. This option is mutually exclusive with `--environment`; omitting both preserves guest defaults. | diff --git a/scripts/nvx_tools/control_session.py b/scripts/nvx_tools/control_session.py index 624d6188d..d3d25052c 100644 --- a/scripts/nvx_tools/control_session.py +++ b/scripts/nvx_tools/control_session.py @@ -2,12 +2,15 @@ from __future__ import annotations +import contextlib import errno +import math import os import secrets import socket import struct import time +from collections.abc import Generator from dataclasses import dataclass from pathlib import Path from typing import Any, cast @@ -184,49 +187,132 @@ def read_exact(self, length: int, deadline: float) -> bytes: output.extend(chunk) return bytes(output) - def write_all(self, data: bytes) -> None: - try: - self._connection.sendall(data) - except _PEER_CLOSED as error: - raise ControlEndpointClosed() from error + def write_all(self, data: bytes, deadline: float) -> None: + unsent = memoryview(data) + while unsent: + remaining = remaining_timeout(deadline) + if remaining <= 0: + raise TimeoutError("managed control request timed out") + self._connection.settimeout(min(remaining, 0.25)) + try: + sent = self._connection.send(unsent) + except TimeoutError: + continue + except _PEER_CLOSED as error: + raise ControlEndpointClosed() from error + if not sent: + raise ControlEndpointClosed() + unsent = unsent[sent:] def close(self) -> None: self._connection.close() if os.name == "nt": + import _winapi import ctypes - import msvcrt - - _kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) - _peek_named_pipe = _kernel32.PeekNamedPipe - _peek_named_pipe.argtypes = [ - ctypes.c_void_p, - ctypes.c_void_p, - ctypes.c_uint32, - ctypes.c_void_p, - ctypes.POINTER(ctypes.c_uint32), - ctypes.c_void_p, - ] - _peek_named_pipe.restype = ctypes.c_int + + _cancel_io_ex = ctypes.WinDLL("kernel32", use_last_error=True).CancelIoEx + _cancel_io_ex.argtypes = [ctypes.c_void_p, ctypes.c_void_p] + _cancel_io_ex.restype = ctypes.c_int else: + _winapi = cast(Any, None) ctypes = cast(Any, None) - msvcrt = cast(Any, None) - _peek_named_pipe = cast(Any, None) + _cancel_io_ex = cast(Any, None) + +# Windows fails a transfer on a pipe that its server closed or disconnected with +# ERROR_BROKEN_PIPE, ERROR_NO_DATA, or ERROR_PIPE_NOT_CONNECTED. +_PIPE_CLOSED_ERRORS = frozenset((109, 232, 233)) +_ERROR_NOT_FOUND = 1168 + + +@contextlib.contextmanager +def _pipe_closure_reported() -> Generator[None]: + try: + yield + except OSError as error: + if getattr(error, "winerror", None) in _PIPE_CLOSED_ERRORS: + raise ControlEndpointClosed() from error + raise + + +def _signaled_before(event: int, deadline: float) -> bool: + while True: + remaining = remaining_timeout(deadline) + if remaining <= 0: + return False + try: + # On the main thread, Ctrl+C ends this wait too. + result = _winapi.WaitForMultipleObjects( + [event], False, math.ceil(min(remaining, 0.25) * 1000) + ) + except InterruptedError: + continue + if result == _winapi.WAIT_OBJECT_0: + return True + + +def _cancel_transfers(handle: int) -> None: + # Overlapped.cancel() marks a transfer as ended at once, so that releasing the + # transfer before GetOverlappedResult() reports its end would free a buffer + # that Windows may still use. CancelIoEx() leaves the transfer pending, and + # releasing a pending transfer waits for its end or leaks its buffer. + if not _cancel_io_ex(handle, None): + error = ctypes.get_last_error() + if error != _ERROR_NOT_FOUND: + raise ctypes.WinError(error) + + +def _finish_transfer(handle: int, transfer: Any, deadline: float, message: str) -> int: + """Returns how many bytes an overlapped pipe transfer moved. + + A transfer that is still pending at `deadline` is cancelled, and fails with a + TimeoutError unless it completed first. Windows uses the transfer's buffer + until the transfer ends, so this waits for that even when it fails. + """ + try: + completed = _signaled_before(transfer.event, deadline) + except BaseException: + with contextlib.suppress(OSError): + _cancel_transfers(handle) + transfer.GetOverlappedResult(True) + raise + if not completed: + _cancel_transfers(handle) + count, status = transfer.GetOverlappedResult(True) + if status == _winapi.ERROR_OPERATION_ABORTED: + raise TimeoutError(message) + return count class _NamedPipeStream: - def __init__(self, fd: int) -> None: - self._fd = fd + """A named-pipe client whose transfers end by their deadlines. + + A synchronous write blocks for as long as the endpoint does not drain the + pipe, so the client opens the pipe for overlapped I/O and cancels a transfer + that outlives its deadline. + """ + + def __init__(self, handle: int) -> None: + self._handle = handle @classmethod def connect(cls, path: Path, timeout: float) -> _NamedPipeStream: normalized = os.fspath(path).replace("/", "\\") deadline = time.monotonic() + timeout - flags = os.O_RDWR | getattr(os, "O_BINARY", 0) while True: try: - return cls(os.open(normalized, flags)) + return cls( + _winapi.CreateFile( + normalized, + _winapi.GENERIC_READ | _winapi.GENERIC_WRITE, + 0, + _winapi.NULL, + _winapi.OPEN_EXISTING, + _winapi.FILE_FLAG_OVERLAPPED, + _winapi.NULL, + ) + ) except OSError as error: if time.monotonic() >= deadline: raise TimeoutError( @@ -237,66 +323,59 @@ def connect(cls, path: Path, timeout: float) -> _NamedPipeStream: errno.EACCES, errno.EBUSY, errno.EAGAIN, - # The Windows CRT reports the reconnect transition as EINVAL. + # The reconnect transition fails with ERROR_PIPE_BUSY, which + # Python reports as EINVAL. errno.EINVAL, ): raise time.sleep(0.025) - def _available(self) -> int: - available = ctypes.c_uint32() - handle = msvcrt.get_osfhandle(self._fd) - if not _peek_named_pipe( - handle, - None, - 0, - None, - ctypes.byref(available), - None, - ): - error = ctypes.get_last_error() - if error in (109, 233): - raise ControlEndpointClosed() - raise OSError(error, "PeekNamedPipe failed") - return int(available.value) - def read_exact(self, length: int, deadline: float) -> bytes: output = bytearray() while len(output) != length: - if time.monotonic() >= deadline: + if remaining_timeout(deadline) <= 0: raise TimeoutError("managed control response timed out") - available = self._available() - if available == 0: - time.sleep(0.01) - continue - chunk = os.read(self._fd, min(length - len(output), available)) + with _pipe_closure_reported(): + transfer, _ = _winapi.ReadFile( + self._handle, length - len(output), overlapped=True + ) + _finish_transfer( + self._handle, + transfer, + deadline, + "managed control response timed out", + ) + chunk = transfer.getbuffer() if not chunk: raise ControlEndpointClosed() output.extend(chunk) return bytes(output) - def write_all(self, data: bytes) -> None: - remaining = memoryview(data) - while remaining: - try: - count = os.write(self._fd, remaining) - except OSError as error: - # The CRT reports a write to a pipe that the server closed as - # EINVAL, so report it as a closure, as reads do. - if error.errno == errno.EINVAL: - self._available() - raise - if count <= 0: + def write_all(self, data: bytes, deadline: float) -> None: + unsent = memoryview(data) + while unsent: + if remaining_timeout(deadline) <= 0: + raise TimeoutError("managed control request timed out") + with _pipe_closure_reported(): + transfer, _ = _winapi.WriteFile(self._handle, unsent, overlapped=True) + count = _finish_transfer( + self._handle, + transfer, + deadline, + "managed control request timed out", + ) + if not count: raise ControlEndpointClosed() - remaining = remaining[count:] + unsent = unsent[count:] def close(self) -> None: - os.close(self._fd) + _winapi.CloseHandle(self._handle) class ControlSession: def __init__(self, stream: _SocketStream | _NamedPipeStream) -> None: self._stream = stream + self._closed = False self._instance_id = bytes(16) self._epoch = 0 self._send_sequence = 0 @@ -318,14 +397,15 @@ def connect( ) session = cls(stream) try: + deadline = time.monotonic() + timeout session._write_outer( OUTER_HOST_ATTACH, bytes(16), 0, 0, capability, + deadline, ) - deadline = time.monotonic() + timeout while True: record_type, instance_id, epoch, sequence, payload = ( session._read_outer(deadline) @@ -347,7 +427,22 @@ def connect( session._receive_sequence = sequence + 1 return session except BaseException: - stream.close() + session.close() + raise + + @contextlib.contextmanager + def _transferring_record(self) -> Generator[None]: + """Closes the session if sending or receiving a record fails. + + A failed transfer may leave a record partly sent or partly received, after + which later records would not frame correctly. + """ + if self._closed: + raise ValueError("managed control session is closed") + try: + yield + except BaseException: + self.close() raise def _write_outer( @@ -357,6 +452,7 @@ def _write_outer( epoch: int, sequence: int, payload: bytes, + deadline: float, ) -> None: if len(payload) > OUTER_MAX_PAYLOAD: raise ValueError("control payload exceeds the outer protocol limit") @@ -370,25 +466,33 @@ def _write_outer( sequence, len(payload), ) - self._stream.write_all(header + payload) + with self._transferring_record(): + self._stream.write_all(header + payload, deadline) def _read_outer( self, deadline: float, ) -> tuple[int, bytes, int, int, bytes]: - header = self._stream.read_exact(OUTER_HEADER.size, deadline) - magic, version, record_type, flags, instance_id, epoch, sequence, length = ( - OUTER_HEADER.unpack(header) - ) - if magic != b"NVXS" or version != 1 or flags != 0 or length > OUTER_MAX_PAYLOAD: - raise ScriptError("control endpoint returned an invalid outer record") - payload = self._stream.read_exact(length, deadline) if length else b"" + with self._transferring_record(): + header = self._stream.read_exact(OUTER_HEADER.size, deadline) + magic, version, record_type, flags, instance_id, epoch, sequence, length = ( + OUTER_HEADER.unpack(header) + ) + if ( + magic != b"NVXS" + or version != 1 + or flags != 0 + or length > OUTER_MAX_PAYLOAD + ): + raise ScriptError("control endpoint returned an invalid outer record") + payload = self._stream.read_exact(length, deadline) if length else b"" return record_type, instance_id, epoch, sequence, payload def _send_app( self, kind: int, request_id: int, + deadline: float, payload: bytes = b"", ) -> None: frame = ( @@ -409,6 +513,7 @@ def _send_app( self._epoch, self._send_sequence, frame, + deadline, ) self._send_sequence += 1 @@ -442,8 +547,9 @@ def _request_id() -> int: def ping(self, timeout: float) -> None: request_id = self._request_id() - self._send_app(APP_PING, request_id) - kind, response_id, status, payload = self._read_app(time.monotonic() + timeout) + deadline = time.monotonic() + timeout + self._send_app(APP_PING, request_id, deadline) + kind, response_id, status, payload = self._read_app(deadline) if kind != APP_READY or response_id != request_id or status != 0 or payload: raise ScriptError("managed guest did not acknowledge readiness") @@ -531,10 +637,12 @@ def exec( raise ValueError("managed exec request exceeds the protocol limit") request_id = self._request_id() - self._send_app(APP_EXEC, request_id, payload) + # The response timeout bounds the whole exchange, so an endpoint that + # stops reading cannot hold up sending the request either. + deadline = time.monotonic() + response_timeout + self._send_app(APP_EXEC, request_id, deadline, payload) stdout = bytearray() stderr = bytearray() - deadline = time.monotonic() + response_timeout while True: kind, response_id, status, response = self._read_app(deadline) if response_id != request_id: @@ -567,13 +675,16 @@ def exec( def stop(self, timeout: float) -> None: request_id = self._request_id() - self._send_app(APP_STOP, request_id) - kind, response_id, status, payload = self._read_app(time.monotonic() + timeout) + deadline = time.monotonic() + timeout + self._send_app(APP_STOP, request_id, deadline) + kind, response_id, status, payload = self._read_app(deadline) if kind != APP_STOPPED or response_id != request_id or status != 0 or payload: raise ScriptError("managed guest did not acknowledge stop") def close(self) -> None: - self._stream.close() + if not self._closed: + self._closed = True + self._stream.close() def __enter__(self) -> ControlSession: return self diff --git a/scripts/test_control_session.py b/scripts/test_control_session.py index 87db7f20e..da11964db 100644 --- a/scripts/test_control_session.py +++ b/scripts/test_control_session.py @@ -10,10 +10,16 @@ import unittest import uuid from pathlib import Path +from types import SimpleNamespace +from typing import Any, cast +from unittest.mock import MagicMock, patch sys.path.insert(0, str(Path(__file__).parent)) from nvx_tools import control_session # noqa: E402 +if sys.platform == "win32": + import _winapi + def _read_like_openvmm(descriptor: int) -> bytes | None: """Reads a capability pipe as OpenVMM reads --microvm-control-auth-stdin. @@ -78,8 +84,7 @@ def _read_outer(connection: socket.socket): return (*values[:-1], payload) -def _write_app( - connection: socket.socket, +def _app_record( *, instance_id: bytes, sequence: int, @@ -87,7 +92,7 @@ def _write_app( request_id: int, status: int, payload: bytes, -) -> None: +) -> bytes: app = ( control_session.APP_HEADER.pack( b"NVXC", @@ -100,7 +105,7 @@ def _write_app( ) + payload ) - connection.sendall( + return ( control_session.OUTER_HEADER.pack( b"NVXS", 1, @@ -115,6 +120,105 @@ def _write_app( ) +def _write_app( + connection: socket.socket, + *, + instance_id: bytes, + sequence: int, + kind: int, + request_id: int, + status: int, + payload: bytes, +) -> None: + connection.sendall( + _app_record( + instance_id=instance_id, + sequence=sequence, + kind=kind, + request_id=request_id, + status=status, + payload=payload, + ) + ) + + +class _RecordingStream: + """A transport that replays `responses` and records every transfer.""" + + def __init__(self, responses: bytes = b"", *, failure: str | None = None): + self._responses = bytearray(responses) + # The kind of transfer, "write" or "read", that times out. + self._failure = failure + self.transfers: list[tuple[str, float]] = [] + self.closes = 0 + + def _transfer(self, kind: str, deadline: float) -> None: + self.transfers.append((kind, deadline)) + if kind == self._failure: + raise TimeoutError(f"{kind} timed out") + + def write_all(self, data: bytes, deadline: float) -> None: + self._transfer("write", deadline) + + def read_exact(self, length: int, deadline: float) -> bytes: + self._transfer("read", deadline) + data = bytes(self._responses[:length]) + del self._responses[:length] + return data + + def close(self) -> None: + self.closes += 1 + + +if sys.platform == "win32": + + def _create_pipe(quota: int = 65536) -> tuple[Path, int]: + name = rf"\\.\pipe\nvx-control-session-test-{uuid.uuid4().hex}" + server = _winapi.CreateNamedPipe( + name, + _winapi.PIPE_ACCESS_DUPLEX, + _winapi.PIPE_WAIT, + 1, + quota, + quota, + 0, + _winapi.NULL, + ) + return Path(name), server + + def _accept_client(server: int) -> None: + try: + _winapi.ConnectNamedPipe(server) + except OSError as error: + if error.winerror != _winapi.ERROR_PIPE_CONNECTED: + raise + + def _pipe_read_exact(handle: int, length: int) -> bytes: + output = bytearray() + while len(output) != length: + chunk, _ = _winapi.ReadFile(handle, length - len(output)) + output.extend(chunk) + return bytes(output) + + def _accept_attach(server: int, capability: bytes, instance_id: bytes) -> None: + """Authenticates the client of a pipe as OpenVMM's control endpoint does.""" + _accept_client(server) + header = control_session.OUTER_HEADER.unpack( + _pipe_read_exact(server, control_session.OUTER_HEADER.size) + ) + if ( + header[:4] != (b"NVXS", 1, control_session.OUTER_HOST_ATTACH, 0) + or _pipe_read_exact(server, header[-1]) != capability + ): + raise RuntimeError("test endpoint received an invalid attach record") + _winapi.WriteFile( + server, + control_session.OUTER_HEADER.pack( + b"NVXS", 1, control_session.OUTER_READY, 0, instance_id, 1, 0, 0 + ), + ) + + class ControlSessionTests(unittest.TestCase): def test_exec_streams_output_and_returns_bounded_status(self): client, server = socket.socketpair() @@ -624,6 +728,89 @@ def test_session_reset_is_not_a_closed_endpoint(self): raised.exception, control_session.ControlEndpointClosed ) + def test_operations_bound_their_requests_by_their_deadlines(self): + # Issue #464: an operation's timeout bounds sending its request as well as + # receiving its response, so attach, ping, exec, and stop each write with + # the deadline with which they then read. + instance = bytes.fromhex("66" * 16) + request_id = 0x1234 + + def response(sequence: int, kind: int, payload: bytes = b"") -> bytes: + return _app_record( + instance_id=instance, + sequence=sequence, + kind=kind, + request_id=request_id, + status=0, + payload=payload, + ) + + stream = _RecordingStream( + control_session.OUTER_HEADER.pack( + b"NVXS", 1, control_session.OUTER_READY, 0, instance, 1, 0, 0 + ) + + response(1, control_session.APP_READY) + + response(2, control_session.APP_EXIT, b"exit") + + response(3, control_session.APP_STOPPED) + ) + transport = ( + control_session._NamedPipeStream + if os.name == "nt" + else control_session._SocketStream + ) + with ( + patch.object(transport, "connect", return_value=stream), + patch.object( + control_session, "time", SimpleNamespace(monotonic=lambda: 1000.0) + ), + patch.object( + control_session.ControlSession, "_request_id", return_value=request_id + ), + ): + session = control_session.ControlSession.connect( + Path("endpoint"), bytes(range(1, 33)), 7 + ) + session.ping(5) + session.exec(("/bin/true",), timeout_ms=0, response_timeout=3) + session.stop(2) + session.close() + + self.assertEqual( + stream.transfers, + [ + ("write", 1007.0), + ("read", 1007.0), + *( + transfer + for deadline in (1005.0, 1003.0, 1002.0) + for transfer in ( + ("write", deadline), + ("read", deadline), + ("read", deadline), + ) + ), + ], + ) + + def test_failed_transfer_closes_the_session(self): + # A transfer that fails, as by timing out, may cut its record short, after + # which the endpoint would frame every later record wrongly. + for failure in ("write", "read"): + with self.subTest(failure=failure): + stream = _RecordingStream(failure=failure) + session = control_session.ControlSession(cast(Any, stream)) + with self.assertRaisesRegex(TimeoutError, f"^{failure} timed out$"): + session.ping(5) + self.assertEqual(stream.closes, 1) + transfers = len(stream.transfers) + with self.assertRaisesRegex( + ValueError, "^managed control session is closed$" + ): + session.exec(("/bin/true",), timeout_ms=0, response_timeout=5) + session.close() + self.assertEqual(len(stream.transfers), transfers) + self.assertEqual(stream.closes, 1) + class CapabilityPipeTests(unittest.TestCase): CAPABILITY = bytes(range(1, 33)) @@ -656,6 +843,17 @@ def test_invalid_capabilities_are_rejected(self): class SocketStreamTests(unittest.TestCase): + def test_write_that_sends_nothing_reports_a_closed_endpoint(self): + connection = MagicMock(spec=socket.socket) + connection.send.return_value = 0 + stream = control_session._SocketStream(connection) + with self.assertRaisesRegex( + control_session.ControlEndpointClosed, + "^managed control endpoint closed$", + ): + stream.write_all(b"record", time.monotonic() + 60) + connection.send.assert_called_once() + def test_read_reports_a_closed_endpoint(self): client, server = socket.socketpair() stream = control_session._SocketStream(client) @@ -677,20 +875,89 @@ def test_reset_and_broken_pipe_report_a_closed_endpoint(self): self.skipTest("Unix-domain socket closure semantics require Linux") client, server = socket.socketpair() stream = control_session._SocketStream(client) - stream.write_all(b"unread request") + stream.write_all(b"unread request", time.monotonic() + 5) server.close() try: with self.assertRaises(control_session.ControlEndpointClosed) as raised: stream.read_exact(1, time.monotonic() + 5) self.assertIsInstance(raised.exception.__cause__, ConnectionResetError) with self.assertRaises(control_session.ControlEndpointClosed) as raised: - stream.write_all(b"record") + stream.write_all(b"record", time.monotonic() + 5) self.assertIsInstance(raised.exception.__cause__, BrokenPipeError) finally: stream.close() + def test_request_to_a_stalled_endpoint_times_out(self): + # Issue #464: a request that the endpoint does not read must not hold the + # operation past its timeout. + if sys.platform != "linux": + self.skipTest("Unix-domain socket buffering semantics require Linux") + client, server = socket.socketpair() + # The request is several times larger than the socket buffer. + client.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 4096) + server.settimeout(5) + session = control_session.ControlSession(control_session._SocketStream(client)) + session._instance_id = bytes.fromhex("77" * 16) + session._epoch = 1 + arguments = ("/bin/true", *("a" * 4000,) * 15) + released = threading.Event() + received = bytearray() + failures: list[BaseException] = [] + + def serve() -> None: + try: + # The peer reads nothing until the client gives up, or long + # after the client's timeout. + released.wait(10) + while chunk := server.recv(65536): + received.extend(chunk) + except BaseException as error: + failures.append(error) + + worker = threading.Thread(target=serve, daemon=True) + worker.start() + timeout = 0.25 + try: + started = time.monotonic() + with self.assertRaisesRegex( + TimeoutError, "^managed control request timed out$" + ): + session.exec(arguments, timeout_ms=0, response_timeout=timeout) + elapsed = time.monotonic() - started + with self.assertRaisesRegex( + ValueError, "^managed control session is closed$" + ): + session.ping(5) + finally: + session.close() + released.set() + worker.join(10) + server.close() + + self.assertFalse(worker.is_alive()) + self.assertEqual(failures, []) + self.assertGreaterEqual(elapsed, timeout) + self.assertLess(elapsed, timeout + 2) + # The endpoint received only part of the request before the end of stream. + self.assertLess(len(received), sum(map(len, arguments))) + class NamedPipeStreamTests(unittest.TestCase): + def test_write_that_moves_nothing_reports_a_closed_endpoint(self): + # The stream is never closed, since its handle is not real. + stream = control_session._NamedPipeStream(123) + with ( + patch.object(control_session, "_winapi") as winapi, + patch.object(control_session, "_finish_transfer", return_value=0) as finish, + self.assertRaisesRegex( + control_session.ControlEndpointClosed, + "^managed control endpoint closed$", + ), + ): + winapi.WriteFile.return_value = (MagicMock(), 0) + stream.write_all(b"record", time.monotonic() + 60) + finish.assert_called_once() + def test_write_reports_a_closed_endpoint(self): # OpenVMM closes the pipe as it tears the VM down, and the CRT reports a # later write to it as EINVAL rather than as a broken pipe. @@ -718,10 +985,126 @@ def test_write_reports_a_closed_endpoint(self): control_session.ControlEndpointClosed, "^managed control endpoint closed$", ): - stream.write_all(b"record") + stream.write_all(b"record", time.monotonic() + 5) + finally: + stream.close() + + def test_read_reports_a_closed_endpoint_after_its_data(self): + if sys.platform != "win32": + self.skipTest("named pipes require a Windows host") + path, server = _create_pipe() + try: + stream = control_session._NamedPipeStream.connect(path, 5) + _accept_client(server) + _winapi.WriteFile(server, b"ok") + finally: + _winapi.CloseHandle(server) + try: + self.assertEqual(stream.read_exact(2, time.monotonic() + 5), b"ok") + with self.assertRaisesRegex( + control_session.ControlEndpointClosed, + "^managed control endpoint closed$", + ): + stream.read_exact(1, time.monotonic() + 5) finally: stream.close() + def test_read_times_out_without_losing_later_data(self): + if sys.platform != "win32": + self.skipTest("named pipes require a Windows host") + path, server = _create_pipe() + self.addCleanup(_winapi.CloseHandle, server) + stream = control_session._NamedPipeStream.connect(path, 5) + self.addCleanup(stream.close) + _accept_client(server) + + started = time.monotonic() + with self.assertRaisesRegex( + TimeoutError, "^managed control response timed out$" + ): + stream.read_exact(2, started + 0.1) + self.assertLess(time.monotonic() - started, 2.1) + _winapi.WriteFile(server, b"ok") + self.assertEqual(stream.read_exact(2, time.monotonic() + 5), b"ok") + + def test_interrupted_write_is_cancelled_before_the_interruption_propagates(self): + if sys.platform != "win32": + self.skipTest("named pipes require a Windows host") + path, server = _create_pipe(1024) + self.addCleanup(_winapi.CloseHandle, server) + stream = control_session._NamedPipeStream.connect(path, 5) + self.addCleanup(stream.close) + _accept_client(server) + + with ( + patch.object( + control_session, "_signaled_before", side_effect=KeyboardInterrupt + ), + self.assertRaises(KeyboardInterrupt), + ): + stream.write_all(b"x" * 4096, time.monotonic() + 5) + # The write, which the pipe could not hold, no longer waits in it. + self.assertEqual(_winapi.PeekNamedPipe(server)[0], 0) + + def test_request_to_a_stalled_endpoint_times_out_and_closes_the_session(self): + # Issue #464: an endpoint that authenticated the session and then stopped + # reading held the request's write, and so the operation, past its timeout. + if sys.platform != "win32": + self.skipTest("named pipes require a Windows host") + # The 4,093-byte request does not fit in the endpoint's 1 KiB quota. + path, server = _create_pipe(1024) + capability = bytes(range(1, 33)) + released = threading.Event() + received = bytearray() + failures: list[BaseException] = [] + + def serve() -> None: + # Type checkers check a nested function apart from its test's skip. + assert sys.platform == "win32" + try: + _accept_attach(server, capability, bytes.fromhex("11" * 16)) + # The endpoint keeps the pipe open but reads nothing more until + # the client gives up, or long after the client's timeout. + released.wait(10) + while True: + chunk, _ = _winapi.ReadFile(server, 65536) + received.extend(chunk) + except BrokenPipeError: + pass + except BaseException as error: + failures.append(error) + + worker = threading.Thread(target=serve, daemon=True) + worker.start() + timeout = 0.25 + try: + with control_session.ControlSession.connect(path, capability, 5) as session: + started = time.monotonic() + with self.assertRaisesRegex( + TimeoutError, "^managed control request timed out$" + ): + session.exec( + ("/bin/true", "a" * 4000), + timeout_ms=0, + response_timeout=timeout, + ) + elapsed = time.monotonic() - started + with self.assertRaisesRegex( + ValueError, "^managed control session is closed$" + ): + session.ping(5) + finally: + released.set() + worker.join(5) + _winapi.CloseHandle(server) + + self.assertFalse(worker.is_alive()) + self.assertEqual(failures, []) + self.assertGreaterEqual(elapsed, timeout) + self.assertLess(elapsed, timeout + 2) + # The endpoint received less than the request before the pipe closed. + self.assertLess(len(received), 4000) + if __name__ == "__main__": unittest.main() diff --git a/scripts/test_microvm_tests.py b/scripts/test_microvm_tests.py index 1d527302b..9d5e6ba80 100644 --- a/scripts/test_microvm_tests.py +++ b/scripts/test_microvm_tests.py @@ -2438,9 +2438,7 @@ class ControlSessionTests(unittest.TestCase): def test_named_pipe_connect_retries_transient_invalid_argument(self): error = OSError(control_session.errno.EINVAL, "Invalid argument") with ( - patch.object( - control_session.os, "open", side_effect=[error, 123] - ) as open_pipe, + patch.object(control_session, "_winapi") as winapi, patch.object( control_session.time, "monotonic", @@ -2448,13 +2446,14 @@ def test_named_pipe_connect_retries_transient_invalid_argument(self): ), patch.object(control_session.time, "sleep") as sleep, ): + winapi.CreateFile.side_effect = [error, 123] stream = control_session._NamedPipeStream.connect( Path(r"\\.\pipe\nvx-test"), 1.0, ) - self.assertEqual(stream._fd, 123) - self.assertEqual(open_pipe.call_count, 2) + self.assertEqual(stream._handle, 123) + self.assertEqual(winapi.CreateFile.call_count, 2) sleep.assert_called_once_with(0.025)