fix: text encoding issues

Author: p-

src/seclab_taskflows/mcp_servers/local_file_viewer.py

@@ -66,7 +66,7 @@ def search_zipfile(database_path, term, search_dir=None):
                 continue
             with z.open(entry, "r") as f:
                 for i, line in enumerate(f):
-                    if term in str(line):
+                    if term in line.decode("utf-8", errors="replace"):
                         filename = remove_root_dir(entry.filename)
                         if not filename in results:
                             results[filename] = [i + 1]
@@ -104,7 +104,7 @@ def get_file(database_path, filename):
                 continue
             if remove_root_dir(entry.filename) == filename:
                 with z.open(entry, "r") as f:
-                    results = [line.rstrip() for line in f]
+                    results = [line.decode("utf-8", errors="replace").rstrip() for line in f]
                     return results
     return results
 

tests/test_local_file_viewer.py

@@ -0,0 +1,87 @@
+# SPDX-FileCopyrightText: GitHub, Inc.
+# SPDX-License-Identifier: MIT
+
+import os
+import tempfile
+import zipfile
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+import seclab_taskflows.mcp_servers.local_file_viewer as lfv_mod
+from seclab_taskflows.mcp_servers.local_file_viewer import get_file, search_zipfile
+
+# The zip entries use a root directory prefix, like a GitHub zipball, which
+# local_file_viewer strips via remove_root_dir().
+ROOT = "owner-repo-abc1234"
+
+SAMPLE = 'import os\nprint("héllo wörld")\n'
+
+
+def _make_zip_on_disk(tmp_dir, owner, repo, files):
+    """Write a zipball-style archive to {tmp_dir}/{owner}/{repo}.zip."""
+    owner_dir = Path(tmp_dir) / owner
+    owner_dir.mkdir(parents=True, exist_ok=True)
+    zip_path = owner_dir / f"{repo}.zip"
+    with zipfile.ZipFile(zip_path, "w") as zf:
+        for path, content in files.items():
+            zf.writestr(f"{ROOT}/{path}", content)
+    return zip_path
+
+
+class TestGetFileDecodesBytes:
+    def test_get_file_returns_decoded_str_lines(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            zip_path = _make_zip_on_disk(tmp_dir, "owner", "repo", {"foo.py": SAMPLE})
+            lines = get_file(zip_path, "foo.py")
+            # Lines must be decoded str, not bytes, and free of b'...' reprs.
+            assert lines == ["import os", 'print("héllo wörld")']
+            assert all(isinstance(line, str) for line in lines)
+
+    def test_get_file_handles_invalid_utf8_without_crashing(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            owner_dir = Path(tmp_dir) / "owner"
+            owner_dir.mkdir(parents=True)
+            zip_path = owner_dir / "repo.zip"
+            with zipfile.ZipFile(zip_path, "w") as zf:
+                zf.writestr(f"{ROOT}/bin.dat", b"good\n\xff\xfe\n")
+            lines = get_file(zip_path, "bin.dat")
+            assert lines[0] == "good"
+            assert all(isinstance(line, str) for line in lines)
+
+
+class TestSearchZipfileDecodesBytes:
+    def test_search_matches_non_ascii_term(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            zip_path = _make_zip_on_disk(tmp_dir, "owner", "repo", {"foo.py": SAMPLE})
+            results = search_zipfile(zip_path, "héllo")
+            assert results == {"foo.py": [2]}
+
+
+class TestFetchFileContentTool:
+    @pytest.mark.asyncio
+    async def test_fetch_file_content_returns_decoded_numbered_lines(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            _make_zip_on_disk(tmp_dir, "owner", "repo", {"foo.py": SAMPLE})
+            with patch.object(lfv_mod, "LOCAL_GH_DIR", tmp_dir):
+                result = await lfv_mod.fetch_file_content(owner="Owner", repo="Repo", path="foo.py")
+            assert "1: import os" in result
+            assert '2: print("héllo wörld")' in result
+            # Ensure no Python bytes reprs leak into the output.
+            assert "b'" not in result
+
+    @pytest.mark.asyncio
+    async def test_get_file_lines_returns_decoded_range(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            _make_zip_on_disk(tmp_dir, "owner", "repo", {"foo.py": SAMPLE})
+            with patch.object(lfv_mod, "LOCAL_GH_DIR", tmp_dir):
+                result = await lfv_mod.get_file_lines(
+                    owner="owner", repo="repo", path="foo.py", start_line=2, length=1
+                )
+            assert result == '2: print("héllo wörld")'
+            assert "b'" not in result
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])