From 310dbe031fdf5effd411590d4cdcb2c49c7a8b4c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Peter=20St=C3=B6ckli?= Date: Thu, 2 Jul 2026 11:52:21 +0000 Subject: [PATCH 1/5] fix: text encoding issues --- .../mcp_servers/local_file_viewer.py | 4 +- tests/test_local_file_viewer.py | 87 +++++++++++++++++++ 2 files changed, 89 insertions(+), 2 deletions(-) create mode 100644 tests/test_local_file_viewer.py diff --git a/src/seclab_taskflows/mcp_servers/local_file_viewer.py b/src/seclab_taskflows/mcp_servers/local_file_viewer.py index 6b65afe..61ca02f 100644 --- a/src/seclab_taskflows/mcp_servers/local_file_viewer.py +++ b/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 diff --git a/tests/test_local_file_viewer.py b/tests/test_local_file_viewer.py new file mode 100644 index 0000000..f71a66d --- /dev/null +++ b/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"]) From 5505c98e495b8b07de4f81268fa8e675bcac7eac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Peter=20St=C3=B6ckli?= Date: Fri, 3 Jul 2026 14:36:33 +0000 Subject: [PATCH 2/5] add test test_search_handles_invalid_utf8_without_crashing --- tests/test_local_file_viewer.py | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/tests/test_local_file_viewer.py b/tests/test_local_file_viewer.py index f71a66d..3dc7f2c 100644 --- a/tests/test_local_file_viewer.py +++ b/tests/test_local_file_viewer.py @@ -1,7 +1,6 @@ # SPDX-FileCopyrightText: GitHub, Inc. # SPDX-License-Identifier: MIT -import os import tempfile import zipfile from pathlib import Path @@ -58,6 +57,18 @@ def test_search_matches_non_ascii_term(self): results = search_zipfile(zip_path, "héllo") assert results == {"foo.py": [2]} + def test_search_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 match\n") + # errors="replace" means the invalid bytes must not raise and the + # searchable term on the same line is still matched. + results = search_zipfile(zip_path, "match") + assert results == {"bin.dat": [2]} + class TestFetchFileContentTool: @pytest.mark.asyncio From ace045ca04fe86435097f39fc17862ff55d291a3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Peter=20St=C3=B6ckli?= Date: Fri, 3 Jul 2026 14:39:43 +0000 Subject: [PATCH 3/5] fix: same issue in gh_file_viewer --- src/seclab_taskflows/mcp_servers/gh_file_viewer.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/seclab_taskflows/mcp_servers/gh_file_viewer.py b/src/seclab_taskflows/mcp_servers/gh_file_viewer.py index 2d5a706..e2435ad 100644 --- a/src/seclab_taskflows/mcp_servers/gh_file_viewer.py +++ b/src/seclab_taskflows/mcp_servers/gh_file_viewer.py @@ -129,7 +129,7 @@ def search_zipfile(database_path, term): 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] From 52f8b3a2d7f77e892653b1af251e6343cc5b7343 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Peter=20St=C3=B6ckli?= Date: Fri, 3 Jul 2026 14:42:25 +0000 Subject: [PATCH 4/5] try fix for mac and windows tests --- tests/test_local_file_viewer.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/tests/test_local_file_viewer.py b/tests/test_local_file_viewer.py index 3dc7f2c..492749d 100644 --- a/tests/test_local_file_viewer.py +++ b/tests/test_local_file_viewer.py @@ -1,6 +1,7 @@ # SPDX-FileCopyrightText: GitHub, Inc. # SPDX-License-Identifier: MIT +import os import tempfile import zipfile from pathlib import Path @@ -75,7 +76,9 @@ class TestFetchFileContentTool: 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): + # Resolve symlinks (e.g. macOS /var -> /private/var) so the patched + # dir matches sanitize_file_path's os.path.realpath comparison. + with patch.object(lfv_mod, "LOCAL_GH_DIR", os.path.realpath(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 @@ -86,7 +89,9 @@ async def test_fetch_file_content_returns_decoded_numbered_lines(self): 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): + # Resolve symlinks (e.g. macOS /var -> /private/var) so the patched + # dir matches sanitize_file_path's os.path.realpath comparison. + with patch.object(lfv_mod, "LOCAL_GH_DIR", os.path.realpath(tmp_dir)): result = await lfv_mod.get_file_lines( owner="owner", repo="repo", path="foo.py", start_line=2, length=1 ) From ea71e253ecf1fcc680b76eb445e826c36375038f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Peter=20St=C3=B6ckli?= Date: Fri, 3 Jul 2026 15:00:13 +0000 Subject: [PATCH 5/5] add tests for gh fileviewer --- tests/test_gh_file_viewer.py | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/tests/test_gh_file_viewer.py b/tests/test_gh_file_viewer.py index 87e0622..8de3688 100644 --- a/tests/test_gh_file_viewer.py +++ b/tests/test_gh_file_viewer.py @@ -381,6 +381,36 @@ def test_search_zipfile(self): assert 2 in results["main.py"] assert "utils.py" not in results + def test_search_zipfile_matches_non_ascii_term(self): + zip_bytes = _make_zip_bytes({"main.py": 'import os\nprint("héllo wörld")\n'}) + # Close the temp file before reopening it: on Windows a NamedTemporaryFile + # is locked and cannot be opened again by zipfile while still open. + with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f: + f.write(zip_bytes) + path = f.name + try: + results = gfv_mod.search_zipfile(path, "héllo") + finally: + os.unlink(path) + assert results == {"main.py": [2]} + + def test_search_zipfile_handles_invalid_utf8_without_crashing(self): + # Invalid UTF-8 bytes must be decoded with errors="replace" rather than + # raising, and a valid term on the same line is still matched. + buf = BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("owner-repo-abc1234/bin.dat", b"good\n\xff\xfe match\n") + # Close the temp file before reopening it: on Windows a NamedTemporaryFile + # is locked and cannot be opened again by zipfile while still open. + with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f: + f.write(buf.getvalue()) + path = f.name + try: + results = gfv_mod.search_zipfile(path, "match") + finally: + os.unlink(path) + assert results == {"bin.dat": [2]} + if __name__ == "__main__": pytest.main([__file__, "-v"])