diff --git a/src/azure-cli/azure/cli/command_modules/containerapp/_ssh_utils.py b/src/azure-cli/azure/cli/command_modules/containerapp/_ssh_utils.py index d5f16cb99b9..828ceb53054 100644 --- a/src/azure-cli/azure/cli/command_modules/containerapp/_ssh_utils.py +++ b/src/azure-cli/azure/cli/command_modules/containerapp/_ssh_utils.py @@ -102,10 +102,34 @@ def recv(self, *args, **kwargs): return self._socket.recv(*args, **kwargs) +def _write_to_terminal(text): + # The terminal's encoding (e.g. cp1252 on Windows) may not be able to + # represent every character the container emits (emoji, non-Latin scripts). + # On a UTF-8 terminal the fast path prints natively; only when the console + # codec cannot encode a character do we fall back to a non-failing policy so + # the rest of the output is still shown instead of crashing the exec session. + try: + print(text, end="", flush=True) + return + except UnicodeEncodeError: + pass + + encoding = getattr(sys.stdout, "encoding", None) or SSH_DEFAULT_ENCODING + encoded = text.encode(encoding, errors="backslashreplace") + buffer = getattr(sys.stdout, "buffer", None) + if buffer is not None: + buffer.write(encoded) + buffer.flush() + else: + # Stream has no binary buffer (already wrapped / replaced) -> round-trip + # through the same codec so the write itself cannot raise. + print(encoded.decode(encoding, errors="backslashreplace"), end="", flush=True) + + def _decode_and_output_to_terminal(connection: WebSocketConnection, response, encodings): for i, encoding in enumerate(encodings): try: - print(response[2:].decode(encoding), end="", flush=True) + decoded = response[2:].decode(encoding) break except UnicodeDecodeError as e: if i == len(encodings) - 1: # ran out of encodings to try @@ -114,7 +138,11 @@ def _decode_and_output_to_terminal(connection: WebSocketConnection, response, en logger.info("Cluster Control Byte: %s", response[1]) logger.info("Hexdump: %s", response[2:].hex()) raise CLIInternalError("Failed to decode server data") from e - logger.info("Failed to encode with encoding %s", encoding) + logger.info("Failed to decode with encoding %s", encoding) + else: + return # empty encodings list: nothing to decode or print + + _write_to_terminal(decoded) def read_ssh(connection: WebSocketConnection, response_encodings): diff --git a/src/azure-cli/azure/cli/command_modules/containerapp/tests/latest/test_ssh_utils.py b/src/azure-cli/azure/cli/command_modules/containerapp/tests/latest/test_ssh_utils.py new file mode 100644 index 00000000000..a250f6a4e22 --- /dev/null +++ b/src/azure-cli/azure/cli/command_modules/containerapp/tests/latest/test_ssh_utils.py @@ -0,0 +1,71 @@ +# -------------------------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for license information. +# -------------------------------------------------------------------------------------------- + +import io +import unittest +from unittest import mock + +from azure.cli.command_modules.containerapp import _ssh_utils +from azure.cli.command_modules.containerapp._ssh_utils import ( + _write_to_terminal, + _decode_and_output_to_terminal, +) + +# Lobster emoji (U+1F99E): valid UTF-8, but has no representation in cp1252. +EMOJI = "\U0001f99e" + + +class _NarrowStdout: + """Mimics a console whose text codec (e.g. cp1252) cannot encode every + character. Its ``write`` re-encodes like a real console, so emoji raises + ``UnicodeEncodeError``; bytes written via the fallback land in ``buffer``.""" + + def __init__(self, encoding="cp1252"): + self.encoding = encoding + self.buffer = io.BytesIO() + + def write(self, text): + text.encode(self.encoding) # raises UnicodeEncodeError on unencodable chars + + def flush(self): + pass + + +class SshUtilsTerminalOutputTest(unittest.TestCase): + def test_write_falls_back_when_char_not_encodable(self): + text = f"hello {EMOJI} world" + stdout = _NarrowStdout(encoding="cp1252") + + with mock.patch.object(_ssh_utils.sys, "stdout", stdout): + _write_to_terminal(text) # must not raise UnicodeEncodeError + + self.assertEqual(stdout.buffer.getvalue(), + text.encode("cp1252", errors="backslashreplace")) + + def test_write_uses_print_fast_path_when_encodable(self): + stdout = _NarrowStdout(encoding="utf-8") + + with mock.patch.object(_ssh_utils.sys, "stdout", stdout): + _write_to_terminal("plain ascii") + + # Encodable text goes through print(); the fallback buffer stays empty. + self.assertEqual(stdout.buffer.getvalue(), b"") + + def test_decode_and_output_handles_emoji_without_disconnecting(self): + payload = f"openclaw {EMOJI} v1".encode("utf-8") + response = bytes([_ssh_utils.SSH_PROXY_FORWARD, + _ssh_utils.SSH_CLUSTER_STDOUT]) + payload + connection = mock.MagicMock() + stdout = _NarrowStdout(encoding="cp1252") + + with mock.patch.object(_ssh_utils.sys, "stdout", stdout): + _decode_and_output_to_terminal(connection, response, ["utf-8", "latin_1"]) + + connection.disconnect.assert_not_called() + self.assertIn(b"openclaw", stdout.buffer.getvalue()) + + +if __name__ == "__main__": + unittest.main()