Merge remote-tracking branch 'origin/main' into stlc/promote-next

Author: kernel-internal[bot]

src/hypeman/lib/__init__.py

@@ -8,13 +8,23 @@
     cp_to_instance_async,
     cp_from_instance_async,
 )
+from ._ws import (
+    SyncWebSocket,
+    AsyncWebSocket,
+    SyncWebSocketConnector,
+    AsyncWebSocketConnector,
+)
 from .exec import ExecResult, ExecProtocolError, exec, exec_async
 
 __all__ = [
+    "AsyncWebSocket",
+    "AsyncWebSocketConnector",
     "CopyCallbacks",
     "CopyProtocolError",
     "ExecProtocolError",
     "ExecResult",
+    "SyncWebSocket",
+    "SyncWebSocketConnector",
     "cp_from_instance",
     "cp_from_instance_async",
     "cp_to_instance",

src/hypeman/lib/cp.py

@@ -150,10 +150,12 @@ def _parse_message(frame: str) -> dict[str, object]:
     return message
 
 
-def _integer_field(message: dict[str, object], name: str, default: int | None = None) -> int:
+def _integer_field(
+    message: dict[str, object], name: str, default: int | None = None, maximum: int | None = None
+) -> int:
     value = message.get(name, default)
-    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
-        raise CopyProtocolError(f"cp header {name} must be a non-negative integer")
+    if isinstance(value, bool) or not isinstance(value, int) or value < 0 or (maximum is not None and value > maximum):
+        raise CopyProtocolError(f"cp header {name} has an invalid integer value")
     return value
 
 
@@ -371,7 +373,7 @@ def _start(self, message: dict[str, object]) -> None:
         if self.header is not None:
             raise CopyProtocolError("cp sent a new header before ending the previous entry")
         target = self._safe_target(message.get("path"))
-        mode = _integer_field(message, "mode")
+        mode = _integer_field(message, "mode", maximum=0o777)
         size = _integer_field(message, "size")
         mtime = _integer_field(message, "mtime")
         uid = _integer_field(message, "uid", 0)
@@ -532,7 +534,7 @@ async def cp_from_instance_async(
     """Asynchronous counterpart to :func:`cp_from_instance`."""
 
     url, headers = connection_settings(client, instance_id, "cp")
-    state = await asyncio.to_thread(_DownloadState, Path(dst_path), archive, callbacks)
+    state = _DownloadState(Path(dst_path), archive, callbacks)
     try:
         async with connector(url, additional_headers=headers, max_size=None) as websocket:
             await websocket.send(_download_request(src_path, follow_symlinks))
@@ -541,7 +543,7 @@ async def cp_from_instance_async(
                     frame = await websocket.recv()
                 except Exception as exc:
                     raise CopyProtocolError("cp download ended before the final marker") from exc
-                await asyncio.to_thread(state.consume, frame)
+                state.consume(frame)
     except BaseException:
-        await asyncio.to_thread(state.abort)
+        state.abort()
         raise

src/hypeman/lib/exec.py

@@ -80,6 +80,8 @@ def _request(
     rows: int | None,
     cols: int | None,
 ) -> _ExecRequest:
+    if isinstance(command, str):
+        raise ValueError("command must be an argument sequence, not a string")
     argv = list(command)
     if not argv:
         raise ValueError("command must contain at least one string argument")
@@ -145,7 +147,9 @@ def exec(
     """Execute a command and collect its merged stdout/stderr bytes.
 
     The request is dispatched once and is never retried. ``stdin`` is sent as binary
-    WebSocket frames. TTY resize tuples are ``(rows, cols)``.
+    WebSocket frames. The protocol has no stdin EOF frame, so commands must stop
+    reading based on their input, another condition, or ``timeout``. TTY resize
+    tuples are ``(rows, cols)``.
     """
 
     request = _request(

tests/lib/test_websocket_lib.py

@@ -468,3 +468,14 @@ def test_cp_download_rejects_malformed_control_frame(tmp_path: Path) -> None:
     connector = FakeConnector(deque([FakeWebSocket(deque(["not-json"]))]))
     with pytest.raises(CopyProtocolError, match="malformed JSON"):
         cp_from_instance(FakeClient(), "inst", "/guest", tmp_path, connector=connector)
+
+
+def test_exec_rejects_a_bare_command_string() -> None:
+    with pytest.raises(ValueError, match="argument sequence"):
+        exec(FakeClient(), "inst", "echo")
+
+
+def test_cp_download_rejects_invalid_mode(tmp_path: Path) -> None:
+    connector = FakeConnector(deque([FakeWebSocket(deque([file_header("file", size=0, mode=0o4777)]))]))
+    with pytest.raises(CopyProtocolError, match="mode"):
+        cp_from_instance(FakeClient(), "inst", "/guest/file", tmp_path, connector=connector)