Avoid UnicodeEncodeError on narrow-encoding output streams (#1251)

bandit/core/manager.py

@@ -2,6 +2,7 @@
 # Copyright 2014 Hewlett-Packard Development Company, L.P.
 #
 # SPDX-License-Identifier: Apache-2.0
+import codecs
 import collections
 import fnmatch
 import io
@@ -174,6 +175,11 @@ def output_results(
 
             formatter = formatters_mgr[output_format]
             report_func = formatter.plugin
+            # Wrap the output file so that unencodable characters (e.g. emoji
+            # in source files on systems with a narrow encoding like GBK) are
+            # escaped rather than causing a UnicodeEncodeError.  This fixes
+            # pre-commit hook failures on non-UTF-8 terminals (issue #1251).
+            output_file = _safe_text_writer(output_file)
             if output_format == "custom":
                 report_func(
                     self,
@@ -367,6 +373,64 @@ def _execute_ast_visitor(self, fname, fdata, data, nosec_lines):
         return score
 
 
+def _safe_text_writer(fileobj):
+    """Wrap a text file object so that unencodable characters are escaped.
+
+    On systems where the terminal or output stream uses a narrow encoding
+    (e.g. GBK on Windows), writing Unicode characters that are outside that
+    encoding raises a ``UnicodeEncodeError``.  This helper re-wraps the stream
+    with ``errors='backslashreplace'`` so that unencodable code-points are
+    rendered as ``\\uXXXX`` / ``\\UXXXXXXXX`` escape sequences instead of
+    crashing.  This fixes pre-commit hook failures for source files that
+    contain emoji or other non-ASCII characters (issue #1251).
+
+    If *fileobj* already supports arbitrary Unicode (e.g. an ``io.StringIO``
+    or a UTF-8 stream), this function returns it unchanged.
+
+    :param fileobj: a writable text file object
+    :return: a writable text file object that will not raise
+             ``UnicodeEncodeError``
+    """
+    encoding = getattr(fileobj, "encoding", None)
+    if encoding is None:
+        # Not a text stream we can introspect — return as-is.
+        return fileobj
+
+    # If the stream's codec cannot represent every Unicode code-point, wrap it.
+    try:
+        codec_info = codecs.lookup(encoding)
+    except LookupError:
+        return fileobj
+
+    # UTF-8 / UTF-16 / UTF-32 cover the full Unicode range — no wrapping needed.
+    if codec_info.name in {
+        "utf-8",
+        "utf-16",
+        "utf-32",
+        "utf-16-le",
+        "utf-16-be",
+        "utf-32-le",
+        "utf-32-be",
+    }:
+        return fileobj
+
+    # Re-wrap only if there is an underlying binary buffer to attach to.
+    if not hasattr(fileobj, "buffer"):
+        return fileobj
+
+    return io.TextIOWrapper(
+        fileobj.buffer,
+        encoding=encoding,
+        errors="backslashreplace",
+        line_buffering=(
+            fileobj.line_buffering
+            if hasattr(fileobj, "line_buffering")
+            else False
+        ),
+        write_through=True,
+    )
+
+
 def _get_files_from_dir(
     files_dir, included_globs=None, excluded_path_strings=None
 ):

tests/unit/core/test_manager.py

@@ -2,6 +2,7 @@
 # Copyright 2015 Hewlett-Packard Development Company, L.P.
 #
 # SPDX-License-Identifier: Apache-2.0
+import io
 import os
 from unittest import mock
 
@@ -220,6 +221,64 @@ def test_output_results_valid_format(self):
             )
         self.assertTrue(os.path.isfile(output_filename))
 
+    def test_safe_text_writer_escapes_unencodable_chars(self):
+        # _safe_text_writer() must re-wrap a narrow-encoding stream so that
+        # characters outside that encoding (e.g. emoji on a GBK terminal) are
+        # rendered as backslash-escape sequences instead of raising
+        # UnicodeEncodeError (issue #1251).
+        buf = io.BytesIO()
+        gbk_stream = io.TextIOWrapper(
+            buf, encoding="gbk", errors="strict", write_through=True
+        )
+
+        safe = manager._safe_text_writer(gbk_stream)
+
+        emoji = "\U0001f31f"
+        # Must not raise even though the emoji is outside GBK's range.
+        safe.write(emoji)
+        safe.flush()
+        buf.seek(0)
+        written = buf.read().decode("gbk", errors="replace")
+        # The emoji must have been replaced by its \\UXXXXXXXX escape sequence.
+        self.assertIn("\\U0001f31f", written)
+        self.assertNotIn(emoji, written)
+
+    def test_safe_text_writer_passthrough_for_utf8(self):
+        # _safe_text_writer() must return the original object unchanged when
+        # the stream already uses a UTF-8 encoding (no wrapping needed).
+        buf = io.BytesIO()
+        utf8_stream = io.TextIOWrapper(
+            buf, encoding="utf-8", errors="strict", write_through=True
+        )
+
+        result = manager._safe_text_writer(utf8_stream)
+        self.assertIs(result, utf8_stream)
+
+    def test_output_results_no_crash_on_narrow_encoding(self):
+        # output_results() must NOT raise when the output file stream uses a
+        # narrow encoding (e.g. GBK) and the report would contain characters
+        # outside that encoding (issue #1251).
+        temp_directory = self.useFixture(fixtures.TempDir()).path
+        output_filename = os.path.join(temp_directory, "_temp_unicode.txt")
+        lines = 5
+        sev_level = constants.LOW
+        conf_level = constants.LOW
+        output_format = "txt"
+
+        with open(
+            output_filename, "w", encoding="gbk", errors="strict"
+        ) as tmp_file:
+            # Must not raise a UnicodeEncodeError or RuntimeError.
+            try:
+                self.manager.output_results(
+                    lines, sev_level, conf_level, tmp_file, output_format
+                )
+            except RuntimeError:
+                self.fail(
+                    "output_results() raised RuntimeError for a GBK stream; "
+                    "_safe_text_writer should have prevented this (issue #1251)"
+                )
+
     @mock.patch("os.path.isdir")
     def test_discover_files_recurse_skip(self, isdir):
         isdir.return_value = True