Fix unicode decode errors.

Author: wxtim

cylc/uiserver/review.py

@@ -29,6 +29,7 @@
 import jinja2
 from jinja2 import select_autoescape
 import json
+import markupsafe
 import mimetypes
 import os
 import pwd
@@ -517,7 +518,7 @@ def suites(
                 continue
             with contextlib.suppress(OSError):
                 data["entries"].append({
-                    "name": str(item),
+                    "name": item.encode().decode('UTF-8', errors='replace'),
                     "info": {},
                     "last_activity_time": (
                         self.get_last_activity_time(user, item))})
@@ -548,8 +549,9 @@ def suites(
                 rosie_info = {}
                 with contextlib.suppress(IOError):
                     for line in open(   # noqa: SIM115
-                        rosie_suite_info, 'r'
+                        rosie_suite_info, 'rb'
                     ).readlines():
+                        line = line.decode('utf-8', errors='replace')
                         if not line.strip().startswith('#') and '=' in line:
                             rosie_key, rosie_val = line.strip().split("=", 1)
                             if rosie_key in ("project", "title"):
@@ -576,8 +578,8 @@ def get_file(self, user, suite, path, path_in_tar=None, mode=None):
             except KeyError:
                 raise cherrypy.HTTPError(404) from None
             f_size = tar_info.size
-            handle = tar_f.extractfile(path_in_tar)
-            if handle.read(2) == "#!":
+            handle = tar_f.extractfile(path_in_tar, 'rb')
+            if self._is_shebang(handle):
                 mime = self.MIME_TEXT_PLAIN
             else:
                 mime = mimetypes.guess_type(
@@ -601,10 +603,10 @@ def get_file(self, user, suite, path, path_in_tar=None, mode=None):
                     return cherrypy.lib.static.serve_file(temp_f.name, mime)
                 finally:
                     temp_f.close()
-            text = handle.read()
+            text = self._get_file_text(handle)
         else:
             f_size = os.stat(f_name).st_size
-            if open(f_name).read(2) == "#!":  # noqa: SIM115
+            if self._is_shebang(f_name):
                 mime = self.MIME_TEXT_PLAIN
             else:
                 mime = mimetypes.guess_type(quote(f_name))[0]
@@ -618,7 +620,7 @@ def get_file(self, user, suite, path, path_in_tar=None, mode=None):
             ):
                 cherrypy.response.headers["Content-Type"] = mime
                 return cherrypy.lib.static.serve_file(f_name, mime)
-            text = open(f_name).read()  # noqa: SIM115
+            text = self._get_file_text(f_name)
         try:
             text = str(text)
             if mode in [None, "text"]:
@@ -627,7 +629,7 @@ def get_file(self, user, suite, path, path_in_tar=None, mode=None):
                 # escape future modifications to this string. In order to
                 # allow log file syntax highlighting (DEBUG, INFO, etc) we
                 # must cast this back to a str to remove this functionality.
-                text = str(jinja2.escape(text))
+                text = str(markupsafe.escape(text))
             lines = text.splitlines()
         except ValueError:
             if path_in_tar:
@@ -1097,3 +1099,28 @@ def _get_user_suite_dir(self, user, suite, *paths):
     def _sort_summary_entries(suite1):
         """Sort suites by last_activity_time."""
         return suite1.get("last_activity_time") or suite1["name"]
+
+    @staticmethod
+    def _has_shebang(file):
+        """File (or handle) has shebang"""
+        # It's a filepath
+        if isinstance(file, str):
+            with open(file, 'rb') as handle:
+                return handle.read(2) == b'!#'
+
+        # It's a file handle: Make sure that you close it later!
+        first_chars = file.read(2)
+        return first_chars in [b'!#', '!#']
+
+    @staticmethod
+    def _get_file_text(file):
+        """File (or handle) has shebang"""
+        if isinstance(file, str):
+            with open(file, 'rb') as handle:
+                return handle.read().decode('utf-8', errors='replace')
+
+        # It's a file handle: Make sure that you close it later!
+        filecontent = file.read()
+        if isinstance(filecontent, bytes):
+            return filecontent.decode('utf-8', errors='replace')
+        return filecontent

tests/unit/test_review.py

@@ -0,0 +1,66 @@
+# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+from io import BytesIO, StringIO
+import pytest
+
+from cylc.uiserver.review import CylcReviewService
+
+
+@pytest.fixture(scope='module')
+def get_garbage_file(tmp_path_factory: pytest.TempdirFactory) -> str:
+    """Create a file containing a byte which cannot be interpreted as UTF-8
+
+    Check that attempting to read this file will lead to an error without
+    extra handling.
+    """
+    fpath = tmp_path_factory.mktemp("temp") / 'garbage'
+    fpath.write_bytes(b'\xe9')
+
+    # Assert that the string is garbage if it's interpreted as utf-8:
+    with pytest.raises(UnicodeDecodeError):
+        fpath.read_text()
+
+    return str(fpath)
+
+
+def test_has_shebang_handle_contains_garbage():
+    """Test is_shebang method for file handle input."""
+    handle = BytesIO(b'\xe9')
+    assert CylcReviewService._has_shebang(handle) is False
+
+
+def test_has_shebang_handle_is_true():
+    """It recognizes shebangs in text and binary files."""
+    handle = StringIO('!#HelloWorld')
+    assert CylcReviewService._has_shebang(handle) is True
+    handle = BytesIO(b'!#HelloWorld')
+    assert CylcReviewService._has_shebang(handle) is True
+
+
+def test_has_shebang_file_bin_bad(get_garbage_file):
+    """Test is_shebang method for file path input.
+
+    Additionally, test a case where there is a non-utf-8 char at the
+    start of the file.
+    """
+    assert CylcReviewService._has_shebang(get_garbage_file) is False
+
+
+def test_get_file_text_nasty(get_garbage_file):
+    """_get_file_text replaces non UTF-8 chars with question marks."""
+    res = CylcReviewService()._get_file_text(get_garbage_file)
+    assert res == '�'

tox.ini

@@ -18,6 +18,7 @@ exclude=
     .eggs,
     cylc/uiserver/jupyter_config.py,
     *.svg,
+    ./src
 paths =
     ./cylc/uiserver
     ./tests/integration