Fix exec_run() returning empty output over SSH connections

podman/api/client.py

@@ -460,3 +460,22 @@ def _request(
             )
         except OSError as e:
             raise APIError(uri.geturl(), explanation=f"{method.upper()} operation failed") from e
+
+    def _get_raw_response_socket(self, response):
+        """Extract the raw socket from an HTTP response.
+
+        After a 101 Upgrade, data flows directly on the socket outside HTTP
+        framing. This method navigates the urllib3/http.client internals to
+        reach the underlying transport socket.
+        """
+        if self.base_url.scheme == "http+ssh":
+            sock = response.raw._fp.fp.raw._sock
+        else:
+            sock = response.raw._fp.fp.raw
+        try:
+            # Keep a reference to the response to prevent garbage collection
+            # from closing the socket prematurely.
+            sock._response = response
+        except AttributeError:
+            pass
+        return sock

podman/api/ssh.py

@@ -104,12 +104,9 @@ def send(self, data: bytes, flags=None) -> int:  # pylint: disable=unused-argume
         Raises:
             RuntimeError: When socket has not been connected.
         """
-        if not self._proc or self._proc.stdin.closed:
+        if not self._proc:
             raise RuntimeError(f"SSHSocket({self.uri}) not connected.")
-
-        count = self._proc.stdin.write(data)
-        self._proc.stdin.flush()
-        return count
+        return super().send(data, flags or 0)
 
     def recv(self, buffersize, flags=None) -> bytes:  # pylint: disable=unused-argument
         """Read data from SSH forwarded UNIX domain socket.
@@ -123,7 +120,7 @@ def recv(self, buffersize, flags=None) -> bytes:  # pylint: disable=unused-argum
         """
         if not self._proc:
             raise RuntimeError(f"SSHSocket({self.uri}) not connected.")
-        return self._proc.stdout.read(buffersize)
+        return super().recv(buffersize, flags or 0)
 
     def close(self):
         """Release resources held by SSHSocket.

podman/domain/containers.py

@@ -2,7 +2,9 @@
 
 import json
 import logging
+import select
 import shlex
+import struct
 from collections.abc import Iterable, Iterator, Mapping
 from contextlib import suppress
 from typing import Any, Optional, Union
@@ -19,6 +21,65 @@
 logger = logging.getLogger("podman.containers")
 
 
+def _read_from_socket(sock, n=4096):
+    """Read at most n bytes from a socket, using select to wait for data."""
+    if not hasattr(select, "poll"):
+        ready, _, _ = select.select([sock], [], [], 5.0)
+        if not ready:
+            return b""
+    else:
+        poll = select.poll()
+        poll.register(sock, select.POLLIN | select.POLLPRI)
+        if not poll.poll(5000):
+            return b""
+    return sock.recv(n)
+
+
+def _read_exactly(sock, n):
+    """Read exactly n bytes from socket."""
+    data = b""
+    while len(data) < n:
+        chunk = _read_from_socket(sock, n - len(data))
+        if not chunk:
+            return data
+        data += chunk
+    return data
+
+
+def _frames_iter(sock):
+    """Yield (stream_type, payload) tuples from a multiplexed socket."""
+    while True:
+        header = _read_exactly(sock, 8)
+        if len(header) < 8:
+            return
+        stream_type, frame_length = struct.unpack_from(">BxxxL", header)
+        if not frame_length:
+            continue
+        payload = _read_exactly(sock, frame_length)
+        if not payload:
+            return
+        yield stream_type, payload
+
+
+def _consume_frames(sock, demux=False):
+    """Read all frames from socket and return the concatenated result.
+
+    If demux is True, returns (stdout_bytes, stderr_bytes) tuple.
+    Otherwise returns all payload concatenated.
+    """
+    if demux:
+        out = [None, None]
+        for stream_type, payload in _frames_iter(sock):
+            idx = 0 if stream_type == 1 else 1
+            if out[idx] is None:
+                out[idx] = payload
+            else:
+                out[idx] += payload
+        return tuple(out)
+
+    return b"".join(payload for _, payload in _frames_iter(sock))
+
+
 class Container(PodmanResource):
     """Details and configuration for a container managed by the Podman service."""
 
@@ -211,21 +272,48 @@ def exec_run(
         response.raise_for_status()
         exec_id = response.json()['Id']
         # start the exec instance, this will store command output
+        # Upgrade headers are required for SSH connections: without them the
+        # server returns 200 with no Content-Length or Transfer-Encoding, and
+        # http.client closes the socket before the multiplexed body can be
+        # read. With Upgrade the server returns 101 and data flows on the raw
+        # socket outside HTTP framing.
         start_resp = self.api.post(
-            f"/exec/{exec_id}/start", data=json.dumps({"Detach": detach, "Tty": tty}), stream=stream
+            f"/exec/{exec_id}/start",
+            data=json.dumps({"Detach": detach, "Tty": tty}),
+            headers={"Connection": "Upgrade", "Upgrade": "tcp"},
+            stream=True,
         )
         start_resp.raise_for_status()
 
-        if stream:
-            return None, api.stream_frames(start_resp, demux=demux)
+        if start_resp.status_code == 101:
+            sock = self.api._get_raw_response_socket(start_resp)
+            if stream:
+                if demux:
+                    def _demux_gen():
+                        for stream_type, payload in _frames_iter(sock):
+                            if stream_type == 1:
+                                yield (payload, None)
+                            else:
+                                yield (None, payload)
+                    return None, _demux_gen()
+                return None, (payload for _, payload in _frames_iter(sock))
+            output = _consume_frames(sock, demux=demux)
+            try:
+                sock.close()
+            except OSError:
+                pass
+            start_resp.close()
+        else:
+            if stream:
+                return None, api.stream_frames(start_resp, demux=demux)
+            output = start_resp.content
+            if demux:
+                output = demux_output(output)
 
         # get and return exec information
         response = self.api.get(f"/exec/{exec_id}/json")
         response.raise_for_status()
-        if demux:
-            stdout_data, stderr_data = demux_output(start_resp.content)
-            return response.json().get('ExitCode'), (stdout_data, stderr_data)
-        return response.json().get('ExitCode'), start_resp.content
+        return response.json().get('ExitCode'), output
 
     def export(self, chunk_size: int = api.DEFAULT_CHUNK_SIZE) -> Iterator[bytes]:
         """Download container's filesystem contents as a tar archive.