From 00f5ba72f2fc0b6b1171335f29dd0b33021b5c74 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B3=AF=E5=B2=B8=E3=80=80=E4=BA=AE?= <1920071390@campus.ouj.ac.jp> Date: Fri, 10 Jul 2026 14:15:34 +0900 Subject: [PATCH 1/4] fix(security): bound resource use and verify Tirith provenance --- hermes_cli/backup.py | 116 +++++++++++++++++---- tests/hermes_cli/test_backup.py | 152 ++++++++++++++++++++++++++-- tests/tools/test_read_extract.py | 27 +++++ tests/tools/test_tirith_security.py | 111 ++++++++++---------- tools/read_extract.py | 28 +++++ tools/tirith_security.py | 49 ++++----- 6 files changed, 369 insertions(+), 114 deletions(-) diff --git a/hermes_cli/backup.py b/hermes_cli/backup.py index 737bea1509a25..8caaf9618eda7 100644 --- a/hermes_cli/backup.py +++ b/hermes_cli/backup.py @@ -131,15 +131,22 @@ # home-relative location on import. Anything not under home is skipped. _EXTERNAL_PREFIX = "_external/" +_MAX_IMPORT_MEMBERS = 100_000 +_MAX_IMPORT_ARCHIVE_BYTES = 2 * 1024 * 1024 * 1024 +_MAX_IMPORT_MEMBER_BYTES = 512 * 1024 * 1024 +_MAX_IMPORT_TOTAL_BYTES = 2 * 1024 * 1024 * 1024 +_MAX_IMPORT_COMPRESSION_RATIO = 1_000 +_IMPORT_COPY_CHUNK_BYTES = 1024 * 1024 -def _collect_memory_provider_external_paths() -> List[Path]: - """Return existing absolute paths the active memory provider stores - outside HERMES_HOME, resolved from config only (no network, no init). + +def _collect_memory_provider_external_paths(*, include_missing: bool = False) -> List[Path]: + """Return absolute paths the active memory provider stores outside HERMES_HOME. Reads ``memory.provider`` from config, loads just that provider, and asks it for ``backup_paths()``. Returns an empty list when no external provider is active or the provider can't be loaded — backup must never fail because - of a flaky plugin. + of a flaky plugin. Restore callers may include declared paths that do not + exist yet on the destination machine. """ try: from plugins.memory import _get_active_memory_provider, load_memory_provider @@ -173,7 +180,7 @@ def _collect_memory_provider_external_paths() -> List[Path]: p = Path(raw).expanduser() except Exception: continue - if not p.exists(): + if not include_missing and not p.exists(): continue try: resolved = p.resolve() @@ -495,6 +502,59 @@ def _validate_backup_zip(zf: zipfile.ZipFile) -> tuple[bool, str]: return True, "" +def _validate_import_members(zf: zipfile.ZipFile) -> tuple[bool, str]: + infos = [info for info in zf.infolist() if not info.is_dir()] + if len(infos) > _MAX_IMPORT_MEMBERS: + return False, f"zip contains too many files (maximum {_MAX_IMPORT_MEMBERS:,})" + + total_bytes = 0 + seen_names: set[str] = set() + for info in infos: + if info.filename in seen_names: + return False, f"zip contains a duplicate file name: {info.filename}" + seen_names.add(info.filename) + if info.file_size > _MAX_IMPORT_MEMBER_BYTES: + return False, f"zip member is too large: {info.filename}" + total_bytes += info.file_size + if total_bytes > _MAX_IMPORT_TOTAL_BYTES: + return False, "zip expands beyond the maximum import size" + if info.file_size and info.compress_size and ( + info.file_size / info.compress_size > _MAX_IMPORT_COMPRESSION_RATIO + ): + return False, f"zip member compression ratio is too high: {info.filename}" + + return True, "" + + +def _copy_zip_member( + zf: zipfile.ZipFile, + member: zipfile.ZipInfo, + target: Path, +) -> None: + with zf.open(member) as src, open(target, "wb") as dst: + shutil.copyfileobj(src, dst, length=_IMPORT_COPY_CHUNK_BYTES) + + +def _normalized_import_member_name(filename: str, prefix: str) -> str: + if prefix and filename.startswith(prefix): + return filename[len(prefix):] + return filename + + +def _is_allowed_external_target(target: Path, allowed_roots: List[Path]) -> bool: + try: + resolved_target = target.resolve() + except OSError: + return False + for root in allowed_roots: + try: + resolved_target.relative_to(root.resolve()) + return True + except (ValueError, OSError): + continue + return False + + def _detect_prefix(zf: zipfile.ZipFile) -> str: """Detect if the zip has a common directory prefix wrapping all entries. @@ -527,6 +587,10 @@ def run_import(args) -> None: print(f"Error: File not found: {zip_path}") sys.exit(1) + if zip_path.stat().st_size > _MAX_IMPORT_ARCHIVE_BYTES: + print("Error: Backup archive exceeds the maximum import size") + sys.exit(1) + if not zipfile.is_zipfile(zip_path): print(f"Error: Not a valid zip file: {zip_path}") sys.exit(1) @@ -540,8 +604,19 @@ def run_import(args) -> None: print(f"Error: {reason}") sys.exit(1) + ok, reason = _validate_import_members(zf) + if not ok: + print(f"Error: {reason}") + sys.exit(1) + prefix = _detect_prefix(zf) - members = [n for n in zf.namelist() if not n.endswith("/")] + members = sorted( + (info for info in zf.infolist() if not info.is_dir()), + key=lambda info: _normalized_import_member_name( + info.filename, + prefix, + ).startswith(_EXTERNAL_PREFIX), + ) file_count = len(members) print(f"Backup contains {file_count} files") @@ -577,14 +652,22 @@ def run_import(args) -> None: restored_external = 0 skipped_runtime: list[str] = [] home_dir = Path.home().resolve() + allowed_external_roots: Optional[List[Path]] = None t0 = time.monotonic() - for member in members: + for member_info in members: + member = member_info.filename + rel = _normalized_import_member_name(member, prefix) + # External memory-provider state captured under the reserved # ``_external/`` arc prefix restores to its original home-relative # location (e.g. ~/.honcho/config.json), NOT under HERMES_HOME. - if member.startswith(_EXTERNAL_PREFIX): - ext_rel = member[len(_EXTERNAL_PREFIX):] + if rel.startswith(_EXTERNAL_PREFIX): + if allowed_external_roots is None: + allowed_external_roots = _collect_memory_provider_external_paths( + include_missing=True, + ) + ext_rel = rel[len(_EXTERNAL_PREFIX):] if not ext_rel: continue target = home_dir / ext_rel @@ -594,10 +677,12 @@ def run_import(args) -> None: except ValueError: errors.append(f" {member}: path traversal blocked") continue + if not _is_allowed_external_target(target, allowed_external_roots): + errors.append(f" {member}: not declared by the active memory provider") + continue try: target.parent.mkdir(parents=True, exist_ok=True) - with zf.open(member) as src, open(target, "wb") as dst: - dst.write(src.read()) + _copy_zip_member(zf, member_info, target) # External provider configs commonly hold credentials. if target.suffix in {".json", ".env", ".conf"} or target.name in _SECRET_FILE_NAMES: try: @@ -612,12 +697,6 @@ def run_import(args) -> None: print(f" {restored}/{file_count} files ...") continue - # Strip prefix if detected - if prefix and member.startswith(prefix): - rel = member[len(prefix):] - else: - rel = member - if not rel: continue @@ -642,8 +721,7 @@ def run_import(args) -> None: try: target.parent.mkdir(parents=True, exist_ok=True) - with zf.open(member) as src, open(target, "wb") as dst: - dst.write(src.read()) + _copy_zip_member(zf, member_info, target) if target.name in _SECRET_FILE_NAMES: os.chmod(target, 0o600) restored += 1 diff --git a/tests/hermes_cli/test_backup.py b/tests/hermes_cli/test_backup.py index 17832746ca3aa..57389fe63f656 100644 --- a/tests/hermes_cli/test_backup.py +++ b/tests/hermes_cli/test_backup.py @@ -3,6 +3,7 @@ import json import os import sqlite3 +import sys import zipfile from argparse import Namespace from pathlib import Path @@ -1221,6 +1222,64 @@ def test_progress_with_many_files(self, tmp_path, monkeypatch): assert (hermes_home / "config.yaml").exists() assert (hermes_home / "sessions" / "s0599.json").exists() + def test_import_rejects_excessive_member_count(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + zip_path = tmp_path / "too-many.zip" + self._make_backup_zip(zip_path, {"config.yaml": "model: test\n", "extra.txt": "x"}) + + import hermes_cli.backup as backup_mod + monkeypatch.setattr(backup_mod, "_MAX_IMPORT_MEMBERS", 1) + + with pytest.raises(SystemExit): + backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) + + def test_import_rejects_excessive_archive_size(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + zip_path = tmp_path / "oversized.zip" + self._make_backup_zip(zip_path, {"config.yaml": "model: test\n"}) + + import hermes_cli.backup as backup_mod + monkeypatch.setattr(backup_mod, "_MAX_IMPORT_ARCHIVE_BYTES", 1) + + with pytest.raises(SystemExit): + backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) + + def test_import_rejects_excessive_compression_ratio(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + zip_path = tmp_path / "ratio.zip" + with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: + zf.writestr("config.yaml", "x" * 10000) + + import hermes_cli.backup as backup_mod + monkeypatch.setattr(backup_mod, "_MAX_IMPORT_COMPRESSION_RATIO", 1) + + with pytest.raises(SystemExit): + backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) + + def test_import_rejects_duplicate_member_names(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + zip_path = tmp_path / "duplicate.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("config.yaml", "model: first\n") + with pytest.warns(UserWarning): + zf.writestr("config.yaml", "model: second\n") + + import hermes_cli.backup as backup_mod + with pytest.raises(SystemExit): + backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) + # --------------------------------------------------------------------------- # Profile restoration tests @@ -1260,12 +1319,14 @@ def test_import_creates_profile_wrappers(self, tmp_path, monkeypatch): assert (hermes_home / "profiles" / "coder" / "config.yaml").exists() assert (hermes_home / "profiles" / "researcher" / "config.yaml").exists() - # Wrapper scripts should be created - assert (wrapper_dir / "coder").exists() - assert (wrapper_dir / "researcher").exists() + wrapper_suffix = ".bat" if sys.platform == "win32" else "" + coder_path = wrapper_dir / f"coder{wrapper_suffix}" + researcher_path = wrapper_dir / f"researcher{wrapper_suffix}" + assert coder_path.exists() + assert researcher_path.exists() # Wrappers should contain the right content - coder_wrapper = (wrapper_dir / "coder").read_text() + coder_wrapper = coder_path.read_text() assert "hermes -p coder" in coder_wrapper def test_import_skips_profile_dirs_without_config(self, tmp_path, monkeypatch): @@ -1290,9 +1351,9 @@ def test_import_skips_profile_dirs_without_config(self, tmp_path, monkeypatch): from hermes_cli.backup import run_import run_import(args) - # Only valid profile should get a wrapper - assert (wrapper_dir / "valid").exists() - assert not (wrapper_dir / "empty").exists() + wrapper_suffix = ".bat" if sys.platform == "win32" else "" + assert (wrapper_dir / f"valid{wrapper_suffix}").exists() + assert not (wrapper_dir / f"empty{wrapper_suffix}").exists() def test_import_without_profiles_module(self, tmp_path, monkeypatch): """Import gracefully handles missing profiles module (fresh install).""" @@ -1858,7 +1919,7 @@ def _spy(src, dst): monkeypatch.setattr(bk, "_safe_copy_db", _spy) snap_id = create_quick_snapshot(hermes_home=hermes_home) # The board db was copied via _safe_copy_db (not raw copy). - assert any(s.endswith("boards/work/kanban.db") for s in called["db"]), called["db"] + assert any(Path(s).parts[-3:] == ("boards", "work", "kanban.db") for s in called["db"]), called["db"] copy = hermes_home / "state-snapshots" / snap_id / "kanban" / "boards" / "work" / "kanban.db" rows = sqlite3.connect(str(copy)).execute("SELECT * FROM tasks").fetchall() assert rows == [("w1", "ship")] @@ -2462,16 +2523,89 @@ def test_import_restores_external_to_home_relative_location(self, tmp_path, monk monkeypatch.setattr(Path, "home", lambda: dst_home) from hermes_cli.backup import run_import + import hermes_cli.backup as backup_mod + monkeypatch.setattr( + backup_mod, + "_collect_memory_provider_external_paths", + lambda **kwargs: [dst_home / ".honcho"], + ) run_import(Namespace(zipfile=str(zip_path), force=True)) restored = dst_home / ".honcho" / "config.json" assert restored.exists() assert restored.read_text() == '{"peer":"bob"}' # Credential-shaped file tightened. - assert (restored.stat().st_mode & 0o777) == 0o600 + if os.name != "nt": + assert (restored.stat().st_mode & 0o777) == 0o600 # External state did NOT leak into HERMES_HOME. assert not (hermes_home / "_external").exists() + def test_import_restores_wrapped_external_state(self, tmp_path, monkeypatch): + dst_home = tmp_path / "dst" + dst_home.mkdir() + hermes_home = dst_home / ".hermes" + + zip_path = tmp_path / "wrapped.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr(".hermes/config.yaml", "model: {}\n") + zf.writestr(".hermes/_external/.honcho/config.json", '{"peer":"bob"}') + + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + monkeypatch.setattr(Path, "home", lambda: dst_home) + + import hermes_cli.backup as backup_mod + monkeypatch.setattr( + backup_mod, + "_collect_memory_provider_external_paths", + lambda **kwargs: [dst_home / ".honcho"], + ) + backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) + + assert (dst_home / ".honcho" / "config.json").read_text() == '{"peer":"bob"}' + assert not (hermes_home / "_external").exists() + + def test_restore_collection_includes_declared_missing_path(self, tmp_path, monkeypatch): + import hermes_cli.backup as backup_mod + import plugins.memory as memory_plugins + + missing = tmp_path / ".honcho" + + class _Provider: + def backup_paths(self): + return [str(missing)] + + monkeypatch.setattr(memory_plugins, "_get_active_memory_provider", lambda: "honcho") + monkeypatch.setattr(memory_plugins, "load_memory_provider", lambda name: _Provider()) + + assert backup_mod._collect_memory_provider_external_paths() == [] + assert backup_mod._collect_memory_provider_external_paths( + include_missing=True, + ) == [missing] + + def test_import_skips_undeclared_external_path(self, tmp_path, monkeypatch): + dst_home = tmp_path / "dst" + dst_home.mkdir() + hermes_home = dst_home / ".hermes" + hermes_home.mkdir() + + zip_path = tmp_path / "backup.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("config.yaml", "model: {}\n") + zf.writestr("_external/.profile", "unexpected") + + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + monkeypatch.setattr(Path, "home", lambda: dst_home) + + import hermes_cli.backup as backup_mod + monkeypatch.setattr( + backup_mod, + "_collect_memory_provider_external_paths", + lambda **kwargs: [], + ) + backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) + + assert not (dst_home / ".profile").exists() + def test_import_blocks_external_path_traversal(self, tmp_path, monkeypatch): """A malicious _external/ member that escapes the home dir is blocked.""" dst_home = tmp_path / "dst" diff --git a/tests/tools/test_read_extract.py b/tests/tools/test_read_extract.py index 3757e03c43b5e..fce198f6579a1 100644 --- a/tests/tools/test_read_extract.py +++ b/tests/tools/test_read_extract.py @@ -175,6 +175,19 @@ def test_missing_document_xml_raises(self): with self.assertRaises(ExtractionError): extract_document_text(p) + def test_rejects_oversized_archive_member(self): + p = os.path.join(self.tmp, "large.docx") + _write_docx(p, self._doc('Text')) + + import tools.read_extract as read_extract + original = read_extract.MAX_OFFICE_MEMBER_BYTES + read_extract.MAX_OFFICE_MEMBER_BYTES = 1 + try: + with self.assertRaises(ExtractionError): + extract_document_text(p) + finally: + read_extract.MAX_OFFICE_MEMBER_BYTES = original + # --------------------------------------------------------------------------- # Excel workbooks (.xlsx) — #10740 @@ -237,6 +250,20 @@ def test_not_a_zip_raises(self): with self.assertRaises(ExtractionError): extract_document_text(p) + def test_rejects_excessive_compression_ratio(self): + p = os.path.join(self.tmp, "ratio.xlsx") + with zipfile.ZipFile(p, "w", zipfile.ZIP_DEFLATED) as z: + z.writestr("xl/workbook.xml", "x" * 10000) + + import tools.read_extract as read_extract + original = read_extract.MAX_OFFICE_COMPRESSION_RATIO + read_extract.MAX_OFFICE_COMPRESSION_RATIO = 1 + try: + with self.assertRaises(ExtractionError): + extract_document_text(p) + finally: + read_extract.MAX_OFFICE_COMPRESSION_RATIO = original + # --------------------------------------------------------------------------- # read_file_tool integration diff --git a/tests/tools/test_tirith_security.py b/tests/tools/test_tirith_security.py index 27202ca63ebd9..e667f13f385cd 100644 --- a/tests/tools/test_tirith_security.py +++ b/tests/tools/test_tirith_security.py @@ -15,7 +15,7 @@ @pytest.fixture(autouse=True) -def _reset_resolved_path(): +def _reset_resolved_path(monkeypatch, request): """Pre-set cached path to skip auto-install in scan tests. Tests that specifically test ensure_installed / resolve behavior reset this to None themselves. @@ -25,6 +25,8 @@ def _reset_resolved_path(): _tirith_mod._install_failure_reason = "" _tirith_mod._crash_count = 0 _tirith_mod._circuit_open = False + if not request.cls or request.cls.__name__ != "TestUnsupportedPlatform": + monkeypatch.setattr(_tirith_mod, "is_platform_supported", lambda: True) yield _tirith_mod._resolved_path = None _tirith_mod._install_thread = None @@ -624,59 +626,41 @@ def test_install_aborts_on_cosign_rejection(self, mock_target, mock_dl, assert path is None assert reason == "cosign_verification_failed" - @patch("tools.tirith_security.tarfile.open") @patch("tools.tirith_security._verify_checksum", return_value=True) @patch("tools.tirith_security.shutil.which", return_value=None) @patch("tools.tirith_security._download_file") @patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin") - def test_install_proceeds_without_cosign(self, mock_target, mock_dl, - mock_which, mock_checksum, - mock_tarfile): - """_install_tirith proceeds with SHA-256 only when cosign is not on PATH.""" + def test_install_aborts_without_cosign(self, mock_target, mock_dl, + mock_which, mock_checksum): from tools.tirith_security import _install_tirith - mock_tar = MagicMock() - mock_tar.__enter__ = MagicMock(return_value=mock_tar) - mock_tar.__exit__ = MagicMock(return_value=False) - mock_tar.getmembers.return_value = [] - mock_tarfile.return_value = mock_tar path, reason = _install_tirith() - # Reaches extraction (no binary in mock archive), but got past cosign assert path is None - assert reason == "binary_not_in_archive" - assert mock_checksum.called # SHA-256 verification ran + assert reason == "cosign_missing" + mock_checksum.assert_not_called() - @patch("tools.tirith_security.tarfile.open") @patch("tools.tirith_security._verify_checksum", return_value=True) @patch("tools.tirith_security._verify_cosign", return_value=None) @patch("tools.tirith_security.shutil.which", return_value="/usr/local/bin/cosign") @patch("tools.tirith_security._download_file") @patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin") - def test_install_proceeds_when_cosign_exec_fails(self, mock_target, mock_dl, - mock_which, mock_cosign, - mock_checksum, mock_tarfile): - """_install_tirith falls back to SHA-256 when cosign exists but fails to execute.""" + def test_install_aborts_when_cosign_exec_fails(self, mock_target, mock_dl, + mock_which, mock_cosign, + mock_checksum): from tools.tirith_security import _install_tirith - mock_tar = MagicMock() - mock_tar.__enter__ = MagicMock(return_value=mock_tar) - mock_tar.__exit__ = MagicMock(return_value=False) - mock_tar.getmembers.return_value = [] - mock_tarfile.return_value = mock_tar path, reason = _install_tirith() assert path is None - assert reason == "binary_not_in_archive" # got past cosign - assert mock_checksum.called + assert reason == "cosign_exec_failed" + mock_checksum.assert_not_called() - @patch("tools.tirith_security.tarfile.open") @patch("tools.tirith_security._verify_checksum", return_value=True) @patch("tools.tirith_security.shutil.which", return_value="/usr/local/bin/cosign") @patch("tools.tirith_security._download_file") @patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin") - def test_install_proceeds_when_cosign_artifacts_missing(self, mock_target, - mock_dl, mock_which, - mock_checksum, mock_tarfile): - """_install_tirith proceeds with SHA-256 when .sig/.pem downloads fail.""" + def test_install_aborts_when_cosign_artifacts_missing(self, mock_target, + mock_dl, mock_which, + mock_checksum): from tools.tirith_security import _install_tirith import urllib.request @@ -685,16 +669,10 @@ def _dl_side_effect(url, dest, timeout=10): raise urllib.request.URLError("404 Not Found") mock_dl.side_effect = _dl_side_effect - mock_tar = MagicMock() - mock_tar.__enter__ = MagicMock(return_value=mock_tar) - mock_tar.__exit__ = MagicMock(return_value=False) - mock_tar.getmembers.return_value = [] - mock_tarfile.return_value = mock_tar - path, reason = _install_tirith() assert path is None - assert reason == "binary_not_in_archive" # got past cosign - assert mock_checksum.called + assert reason == "cosign_artifacts_unavailable" + mock_checksum.assert_not_called() @patch("tools.tirith_security.tarfile.open") @patch("tools.tirith_security._verify_checksum", return_value=True) @@ -747,17 +725,23 @@ def _download(url, dest, timeout=10): with open(checksums, "rb") as src, open(dest, "wb") as dst: dst.write(src.read()) return + if url.endswith(".sig") or url.endswith(".pem"): + with open(dest, "wb") as dst: + dst.write(b"verified-by-mock") + return raise AssertionError(f"unexpected download URL: {url}") return _download @patch("tools.tirith_security._verify_checksum", return_value=True) - @patch("tools.tirith_security.shutil.which", return_value=None) + @patch("tools.tirith_security._verify_cosign", return_value=True) + @patch("tools.tirith_security.shutil.which", return_value="/usr/local/bin/cosign") @patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin") def test_install_extracts_regular_tirith_member(self, mock_target, mock_which, - mock_checksum, tmp_path, monkeypatch): + mock_cosign, mock_checksum, + tmp_path, monkeypatch): """A valid regular-file tirith member is installed as a plain file.""" - del mock_target, mock_which, mock_checksum + del mock_target, mock_which, mock_cosign, mock_checksum from tools.tirith_security import _install_tirith payload = b"#!/bin/sh\nexit 0\n" @@ -780,12 +764,14 @@ def test_install_extracts_regular_tirith_member(self, mock_target, mock_which, assert f.read() == payload @patch("tools.tirith_security._verify_checksum", return_value=True) - @patch("tools.tirith_security.shutil.which", return_value=None) + @patch("tools.tirith_security._verify_cosign", return_value=True) + @patch("tools.tirith_security.shutil.which", return_value="/usr/local/bin/cosign") @patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin") def test_install_rejects_non_regular_tirith_member(self, mock_target, mock_which, - mock_checksum, tmp_path, monkeypatch): + mock_cosign, mock_checksum, + tmp_path, monkeypatch): """Symlink or hardlink tar members must not be installed as tirith.""" - del mock_target, mock_which, mock_checksum + del mock_target, mock_which, mock_cosign, mock_checksum from tools.tirith_security import _install_tirith member = tarfile.TarInfo("bin/tirith") @@ -1174,7 +1160,9 @@ def test_failure_marker_respects_hermes_home(self): from tools.tirith_security import _failure_marker_path with patch.dict(os.environ, {"HERMES_HOME": "/custom/hermes"}): result = _failure_marker_path() - assert result == "/custom/hermes/.tirith-install-failed" + assert os.path.normpath(result) == os.path.normpath( + "/custom/hermes/.tirith-install-failed" + ) def test_conftest_isolation_prevents_real_home_writes(self): """The conftest autouse fixture sets HERMES_HOME; verify it's active.""" @@ -1185,14 +1173,17 @@ def test_conftest_isolation_prevents_real_home_writes(self): def test_get_hermes_home_fallback(self): """Without HERMES_HOME set, falls back to the active OS home.""" from tools.tirith_security import _get_hermes_home - with patch.dict(os.environ, {}, clear=True): - # Remove HERMES_HOME entirely. With HOME also absent, expanduser - # falls back to the account database; compute expected under the - # same environment instead of after patch.dict restores HOME. - os.environ.pop("HERMES_HOME", None) - expected = os.path.join(os.path.expanduser("~"), ".hermes") + env = {} + if os.name == "nt": + env["LOCALAPPDATA"] = r"C:\Users\test\AppData\Local" + with patch.dict(os.environ, env, clear=True): + expected = ( + os.path.join(env["LOCALAPPDATA"], "hermes") + if os.name == "nt" + else os.path.join(os.path.expanduser("~"), ".hermes") + ) result = _get_hermes_home() - assert result == expected + assert os.path.normpath(result) == os.path.normpath(expected) # --------------------------------------------------------------------------- @@ -1457,8 +1448,13 @@ class TestMkdtempOSErrorNoSpace: def test_mkdtemp_oserror_returns_no_space(self): from tools.tirith_security import _install_tirith - with patch("tools.tirith_security.tempfile.mkdtemp", - side_effect=OSError(28, "No space left on device")): + with patch( + "tools.tirith_security._detect_target", + return_value="x86_64-unknown-linux-gnu", + ), patch( + "tools.tirith_security.tempfile.mkdtemp", + side_effect=OSError(28, "No space left on device"), + ): result, reason = _install_tirith(log_failures=False) assert result is None assert reason == "no_space" @@ -1480,7 +1476,10 @@ def test_mkdtemp_oserror_propagates_to_ensure_installed(self): from tools.tirith_security import _resolve_tirith_path, _INSTALL_FAILED _tirith_mod._resolved_path = None - with patch("tools.tirith_security.tempfile.mkdtemp", + with patch( + "tools.tirith_security._detect_target", + return_value="x86_64-unknown-linux-gnu", + ), patch("tools.tirith_security.tempfile.mkdtemp", side_effect=OSError(28, "No space left on device")), \ patch("tools.tirith_security.shutil.which", return_value=None), \ diff --git a/tools/read_extract.py b/tools/read_extract.py index 3607703de6002..5c26cf3385c32 100644 --- a/tools/read_extract.py +++ b/tools/read_extract.py @@ -17,6 +17,10 @@ EXTRACTABLE_EXTENSIONS = frozenset({".ipynb", ".docx", ".xlsx"}) MAX_XLSX_BYTES = 50 * 1024 * 1024 +MAX_OFFICE_MEMBER_BYTES = 32 * 1024 * 1024 +MAX_OFFICE_TOTAL_BYTES = 100 * 1024 * 1024 +MAX_OFFICE_MEMBER_COUNT = 1024 +MAX_OFFICE_COMPRESSION_RATIO = 200 _MAX_XLSX_ROWS_PER_SHEET = 5000 _MAX_XLSX_COLS = 256 @@ -104,9 +108,32 @@ def _zip_xml(zf: zipfile.ZipFile, name: str) -> ET.Element: raise ExtractionError(f"Malformed XML in {name}: {exc}") from exc +def _validate_office_archive(path: str, zf: zipfile.ZipFile) -> None: + try: + if Path(path).stat().st_size > MAX_XLSX_BYTES: + raise ExtractionError("Office document exceeds the compressed size limit") + except OSError as exc: + raise ExtractionError(str(exc)) from exc + + infos = [info for info in zf.infolist() if not info.is_dir()] + if len(infos) > MAX_OFFICE_MEMBER_COUNT: + raise ExtractionError("Office document contains too many archive members") + + total_size = 0 + for info in infos: + if info.file_size > MAX_OFFICE_MEMBER_BYTES: + raise ExtractionError(f"Office document member is too large: {info.filename}") + total_size += info.file_size + if total_size > MAX_OFFICE_TOTAL_BYTES: + raise ExtractionError("Office document expands beyond the extraction limit") + if info.file_size and info.compress_size and info.file_size / info.compress_size > MAX_OFFICE_COMPRESSION_RATIO: + raise ExtractionError(f"Office document member compression ratio is too high: {info.filename}") + + def _extract_docx(path: str) -> str: try: with zipfile.ZipFile(path) as zf: + _validate_office_archive(path, zf) root = _zip_xml(zf, "word/document.xml") except zipfile.BadZipFile as exc: raise ExtractionError(f"Not a valid DOCX: {exc}") from exc @@ -133,6 +160,7 @@ def _extract_docx(path: str) -> str: def _extract_xlsx(path: str) -> str: try: with zipfile.ZipFile(path) as zf: + _validate_office_archive(path, zf) names = set(zf.namelist()) shared = _shared_strings(zf, names) sheets = _workbook_sheets(zf) diff --git a/tools/tirith_security.py b/tools/tirith_security.py index 93509604131d1..6bd87f762794b 100644 --- a/tools/tirith_security.py +++ b/tools/tirith_security.py @@ -421,34 +421,24 @@ def _install_tirith(*, log_failures: bool = True) -> tuple[str | None, str]: log("tirith download failed: %s", exc) return None, "download_failed" - # Cosign provenance verification — preferred but not mandatory. - # When cosign is available, we verify that the release was produced - # by the expected GitHub Actions workflow (full supply chain proof). - # Without cosign, SHA-256 checksum + HTTPS still provides integrity - # and transport-level authenticity. - cosign_verified = False - if shutil.which("cosign"): - try: - _download_file(f"{base_url}/checksums.txt.sig", sig_path) - _download_file(f"{base_url}/checksums.txt.pem", cert_path) - except Exception as exc: - logger.info("cosign artifacts unavailable (%s), proceeding with SHA-256 only", exc) - else: - cosign_result = _verify_cosign(checksums_path, sig_path, cert_path) - if cosign_result is True: - cosign_verified = True - elif cosign_result is False: - # Verification explicitly rejected — abort, the release - # may have been tampered with. - log("tirith install aborted: cosign provenance verification failed") - return None, "cosign_verification_failed" - else: - # None = execution failure (timeout/OSError) — proceed - # with SHA-256 only since cosign itself is broken. - logger.info("cosign execution failed, proceeding with SHA-256 only") - else: - logger.info("cosign not on PATH — installing tirith with SHA-256 verification only " - "(install cosign for full supply chain verification)") + if not shutil.which("cosign"): + log("tirith install aborted: cosign is required for release provenance verification") + return None, "cosign_missing" + + try: + _download_file(f"{base_url}/checksums.txt.sig", sig_path) + _download_file(f"{base_url}/checksums.txt.pem", cert_path) + except Exception as exc: + log("tirith install aborted: cosign artifacts unavailable: %s", exc) + return None, "cosign_artifacts_unavailable" + + cosign_result = _verify_cosign(checksums_path, sig_path, cert_path) + if cosign_result is False: + log("tirith install aborted: cosign provenance verification failed") + return None, "cosign_verification_failed" + if cosign_result is not True: + log("tirith install aborted: cosign could not execute successfully") + return None, "cosign_exec_failed" if not _verify_checksum(archive_path, checksums_path, archive_name): return None, "checksum_failed" @@ -476,8 +466,7 @@ def _install_tirith(*, log_failures: bool = True) -> tuple[str | None, str]: return None, "cross_device_copy_failed" os.chmod(dest, os.stat(dest).st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) - verification = "cosign + SHA-256" if cosign_verified else "SHA-256 only" - logger.info("tirith installed to %s (%s)", dest, verification) + logger.info("tirith installed to %s (cosign + SHA-256)", dest) return dest, "" finally: From a04c8ffdcd8c935579b5e55efb44ab415ed36d0d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B3=AF=E5=B2=B8=E3=80=80=E4=BA=AE?= <1920071390@campus.ouj.ac.jp> Date: Sun, 12 Jul 2026 01:54:57 +0900 Subject: [PATCH 2/4] fix(security): preserve checksum fallback for tirith install --- tests/tools/test_tirith_security.py | 66 +++++++++++++++++------------ tools/tirith_security.py | 53 ++++++++++++++--------- 2 files changed, 72 insertions(+), 47 deletions(-) diff --git a/tests/tools/test_tirith_security.py b/tests/tools/test_tirith_security.py index e667f13f385cd..834e57fb9c1ca 100644 --- a/tests/tools/test_tirith_security.py +++ b/tests/tools/test_tirith_security.py @@ -626,41 +626,55 @@ def test_install_aborts_on_cosign_rejection(self, mock_target, mock_dl, assert path is None assert reason == "cosign_verification_failed" + @patch("tools.tirith_security.tarfile.open") @patch("tools.tirith_security._verify_checksum", return_value=True) @patch("tools.tirith_security.shutil.which", return_value=None) @patch("tools.tirith_security._download_file") @patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin") - def test_install_aborts_without_cosign(self, mock_target, mock_dl, - mock_which, mock_checksum): + def test_install_proceeds_without_cosign(self, mock_target, mock_dl, + mock_which, mock_checksum, + mock_tarfile): from tools.tirith_security import _install_tirith + mock_tar = MagicMock() + mock_tar.__enter__ = MagicMock(return_value=mock_tar) + mock_tar.__exit__ = MagicMock(return_value=False) + mock_tar.getmembers.return_value = [] + mock_tarfile.return_value = mock_tar path, reason = _install_tirith() assert path is None - assert reason == "cosign_missing" - mock_checksum.assert_not_called() + assert reason == "binary_not_in_archive" + assert mock_checksum.called + @patch("tools.tirith_security.tarfile.open") @patch("tools.tirith_security._verify_checksum", return_value=True) @patch("tools.tirith_security._verify_cosign", return_value=None) @patch("tools.tirith_security.shutil.which", return_value="/usr/local/bin/cosign") @patch("tools.tirith_security._download_file") @patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin") - def test_install_aborts_when_cosign_exec_fails(self, mock_target, mock_dl, - mock_which, mock_cosign, - mock_checksum): + def test_install_proceeds_when_cosign_exec_fails(self, mock_target, mock_dl, + mock_which, mock_cosign, + mock_checksum, mock_tarfile): from tools.tirith_security import _install_tirith + mock_tar = MagicMock() + mock_tar.__enter__ = MagicMock(return_value=mock_tar) + mock_tar.__exit__ = MagicMock(return_value=False) + mock_tar.getmembers.return_value = [] + mock_tarfile.return_value = mock_tar path, reason = _install_tirith() assert path is None - assert reason == "cosign_exec_failed" - mock_checksum.assert_not_called() + assert reason == "binary_not_in_archive" + assert mock_checksum.called + @patch("tools.tirith_security.tarfile.open") @patch("tools.tirith_security._verify_checksum", return_value=True) @patch("tools.tirith_security.shutil.which", return_value="/usr/local/bin/cosign") @patch("tools.tirith_security._download_file") @patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin") - def test_install_aborts_when_cosign_artifacts_missing(self, mock_target, - mock_dl, mock_which, - mock_checksum): + def test_install_proceeds_when_cosign_artifacts_missing(self, mock_target, + mock_dl, mock_which, + mock_checksum, mock_tarfile): from tools.tirith_security import _install_tirith import urllib.request @@ -669,10 +683,16 @@ def _dl_side_effect(url, dest, timeout=10): raise urllib.request.URLError("404 Not Found") mock_dl.side_effect = _dl_side_effect + mock_tar = MagicMock() + mock_tar.__enter__ = MagicMock(return_value=mock_tar) + mock_tar.__exit__ = MagicMock(return_value=False) + mock_tar.getmembers.return_value = [] + mock_tarfile.return_value = mock_tar + path, reason = _install_tirith() assert path is None - assert reason == "cosign_artifacts_unavailable" - mock_checksum.assert_not_called() + assert reason == "binary_not_in_archive" + assert mock_checksum.called @patch("tools.tirith_security.tarfile.open") @patch("tools.tirith_security._verify_checksum", return_value=True) @@ -725,23 +745,18 @@ def _download(url, dest, timeout=10): with open(checksums, "rb") as src, open(dest, "wb") as dst: dst.write(src.read()) return - if url.endswith(".sig") or url.endswith(".pem"): - with open(dest, "wb") as dst: - dst.write(b"verified-by-mock") - return raise AssertionError(f"unexpected download URL: {url}") return _download @patch("tools.tirith_security._verify_checksum", return_value=True) - @patch("tools.tirith_security._verify_cosign", return_value=True) - @patch("tools.tirith_security.shutil.which", return_value="/usr/local/bin/cosign") + @patch("tools.tirith_security.shutil.which", return_value=None) @patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin") def test_install_extracts_regular_tirith_member(self, mock_target, mock_which, - mock_cosign, mock_checksum, + mock_checksum, tmp_path, monkeypatch): """A valid regular-file tirith member is installed as a plain file.""" - del mock_target, mock_which, mock_cosign, mock_checksum + del mock_target, mock_which, mock_checksum from tools.tirith_security import _install_tirith payload = b"#!/bin/sh\nexit 0\n" @@ -764,14 +779,13 @@ def test_install_extracts_regular_tirith_member(self, mock_target, mock_which, assert f.read() == payload @patch("tools.tirith_security._verify_checksum", return_value=True) - @patch("tools.tirith_security._verify_cosign", return_value=True) - @patch("tools.tirith_security.shutil.which", return_value="/usr/local/bin/cosign") + @patch("tools.tirith_security.shutil.which", return_value=None) @patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin") def test_install_rejects_non_regular_tirith_member(self, mock_target, mock_which, - mock_cosign, mock_checksum, + mock_checksum, tmp_path, monkeypatch): """Symlink or hardlink tar members must not be installed as tirith.""" - del mock_target, mock_which, mock_cosign, mock_checksum + del mock_target, mock_which, mock_checksum from tools.tirith_security import _install_tirith member = tarfile.TarInfo("bin/tirith") diff --git a/tools/tirith_security.py b/tools/tirith_security.py index 6bd87f762794b..99f2634b93a16 100644 --- a/tools/tirith_security.py +++ b/tools/tirith_security.py @@ -385,10 +385,12 @@ def _extract_tirith_binary(tar: tarfile.TarFile, dest_dir: str, log) -> tuple[st def _install_tirith(*, log_failures: bool = True) -> tuple[str | None, str]: """Download and install tirith to $HERMES_HOME/bin/tirith. - Verifies provenance via cosign and SHA-256 checksum. + Always verifies the SHA-256 checksum. When cosign is available, also + verifies release provenance; unavailable or operationally broken cosign + falls back to checksum verification. Returns (installed_path, failure_reason). On success failure_reason is "". failure_reason is a short tag used by the disk marker to decide if the - failure is retryable (e.g. "cosign_missing" clears when cosign appears). + failure is retryable. """ log = logger.warning if log_failures else logger.debug @@ -421,24 +423,32 @@ def _install_tirith(*, log_failures: bool = True) -> tuple[str | None, str]: log("tirith download failed: %s", exc) return None, "download_failed" - if not shutil.which("cosign"): - log("tirith install aborted: cosign is required for release provenance verification") - return None, "cosign_missing" - - try: - _download_file(f"{base_url}/checksums.txt.sig", sig_path) - _download_file(f"{base_url}/checksums.txt.pem", cert_path) - except Exception as exc: - log("tirith install aborted: cosign artifacts unavailable: %s", exc) - return None, "cosign_artifacts_unavailable" - - cosign_result = _verify_cosign(checksums_path, sig_path, cert_path) - if cosign_result is False: - log("tirith install aborted: cosign provenance verification failed") - return None, "cosign_verification_failed" - if cosign_result is not True: - log("tirith install aborted: cosign could not execute successfully") - return None, "cosign_exec_failed" + cosign_verified = False + if shutil.which("cosign"): + try: + _download_file(f"{base_url}/checksums.txt.sig", sig_path) + _download_file(f"{base_url}/checksums.txt.pem", cert_path) + except Exception as exc: + logger.info( + "cosign artifacts unavailable (%s), proceeding with SHA-256 only", + exc, + ) + else: + cosign_result = _verify_cosign(checksums_path, sig_path, cert_path) + if cosign_result is True: + cosign_verified = True + elif cosign_result is False: + log("tirith install aborted: cosign provenance verification failed") + return None, "cosign_verification_failed" + else: + logger.info( + "cosign execution failed, proceeding with SHA-256 only" + ) + else: + logger.info( + "cosign not on PATH - installing tirith with SHA-256 verification only " + "(install cosign for full supply chain verification)" + ) if not _verify_checksum(archive_path, checksums_path, archive_name): return None, "checksum_failed" @@ -466,7 +476,8 @@ def _install_tirith(*, log_failures: bool = True) -> tuple[str | None, str]: return None, "cross_device_copy_failed" os.chmod(dest, os.stat(dest).st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) - logger.info("tirith installed to %s (cosign + SHA-256)", dest) + verification = "cosign + SHA-256" if cosign_verified else "SHA-256 only" + logger.info("tirith installed to %s (%s)", dest, verification) return dest, "" finally: From ba167531ee14e68a356be318be4ae7838af0f254 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B3=AF=E5=B2=B8=E3=80=80=E4=BA=AE?= <1920071390@campus.ouj.ac.jp> Date: Sun, 12 Jul 2026 20:39:15 +0900 Subject: [PATCH 3/4] fix(security): restore Tirith checksum fallback --- tests/tools/test_tirith_security.py | 55 +++++++++++------------------ tools/tirith_security.py | 30 ++++++++-------- 2 files changed, 36 insertions(+), 49 deletions(-) diff --git a/tests/tools/test_tirith_security.py b/tests/tools/test_tirith_security.py index 834e57fb9c1ca..27202ca63ebd9 100644 --- a/tests/tools/test_tirith_security.py +++ b/tests/tools/test_tirith_security.py @@ -15,7 +15,7 @@ @pytest.fixture(autouse=True) -def _reset_resolved_path(monkeypatch, request): +def _reset_resolved_path(): """Pre-set cached path to skip auto-install in scan tests. Tests that specifically test ensure_installed / resolve behavior reset this to None themselves. @@ -25,8 +25,6 @@ def _reset_resolved_path(monkeypatch, request): _tirith_mod._install_failure_reason = "" _tirith_mod._crash_count = 0 _tirith_mod._circuit_open = False - if not request.cls or request.cls.__name__ != "TestUnsupportedPlatform": - monkeypatch.setattr(_tirith_mod, "is_platform_supported", lambda: True) yield _tirith_mod._resolved_path = None _tirith_mod._install_thread = None @@ -634,6 +632,7 @@ def test_install_aborts_on_cosign_rejection(self, mock_target, mock_dl, def test_install_proceeds_without_cosign(self, mock_target, mock_dl, mock_which, mock_checksum, mock_tarfile): + """_install_tirith proceeds with SHA-256 only when cosign is not on PATH.""" from tools.tirith_security import _install_tirith mock_tar = MagicMock() mock_tar.__enter__ = MagicMock(return_value=mock_tar) @@ -642,9 +641,10 @@ def test_install_proceeds_without_cosign(self, mock_target, mock_dl, mock_tarfile.return_value = mock_tar path, reason = _install_tirith() + # Reaches extraction (no binary in mock archive), but got past cosign assert path is None assert reason == "binary_not_in_archive" - assert mock_checksum.called + assert mock_checksum.called # SHA-256 verification ran @patch("tools.tirith_security.tarfile.open") @patch("tools.tirith_security._verify_checksum", return_value=True) @@ -655,6 +655,7 @@ def test_install_proceeds_without_cosign(self, mock_target, mock_dl, def test_install_proceeds_when_cosign_exec_fails(self, mock_target, mock_dl, mock_which, mock_cosign, mock_checksum, mock_tarfile): + """_install_tirith falls back to SHA-256 when cosign exists but fails to execute.""" from tools.tirith_security import _install_tirith mock_tar = MagicMock() mock_tar.__enter__ = MagicMock(return_value=mock_tar) @@ -664,7 +665,7 @@ def test_install_proceeds_when_cosign_exec_fails(self, mock_target, mock_dl, path, reason = _install_tirith() assert path is None - assert reason == "binary_not_in_archive" + assert reason == "binary_not_in_archive" # got past cosign assert mock_checksum.called @patch("tools.tirith_security.tarfile.open") @@ -675,6 +676,7 @@ def test_install_proceeds_when_cosign_exec_fails(self, mock_target, mock_dl, def test_install_proceeds_when_cosign_artifacts_missing(self, mock_target, mock_dl, mock_which, mock_checksum, mock_tarfile): + """_install_tirith proceeds with SHA-256 when .sig/.pem downloads fail.""" from tools.tirith_security import _install_tirith import urllib.request @@ -691,7 +693,7 @@ def _dl_side_effect(url, dest, timeout=10): path, reason = _install_tirith() assert path is None - assert reason == "binary_not_in_archive" + assert reason == "binary_not_in_archive" # got past cosign assert mock_checksum.called @patch("tools.tirith_security.tarfile.open") @@ -753,8 +755,7 @@ def _download(url, dest, timeout=10): @patch("tools.tirith_security.shutil.which", return_value=None) @patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin") def test_install_extracts_regular_tirith_member(self, mock_target, mock_which, - mock_checksum, - tmp_path, monkeypatch): + mock_checksum, tmp_path, monkeypatch): """A valid regular-file tirith member is installed as a plain file.""" del mock_target, mock_which, mock_checksum from tools.tirith_security import _install_tirith @@ -782,8 +783,7 @@ def test_install_extracts_regular_tirith_member(self, mock_target, mock_which, @patch("tools.tirith_security.shutil.which", return_value=None) @patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin") def test_install_rejects_non_regular_tirith_member(self, mock_target, mock_which, - mock_checksum, - tmp_path, monkeypatch): + mock_checksum, tmp_path, monkeypatch): """Symlink or hardlink tar members must not be installed as tirith.""" del mock_target, mock_which, mock_checksum from tools.tirith_security import _install_tirith @@ -1174,9 +1174,7 @@ def test_failure_marker_respects_hermes_home(self): from tools.tirith_security import _failure_marker_path with patch.dict(os.environ, {"HERMES_HOME": "/custom/hermes"}): result = _failure_marker_path() - assert os.path.normpath(result) == os.path.normpath( - "/custom/hermes/.tirith-install-failed" - ) + assert result == "/custom/hermes/.tirith-install-failed" def test_conftest_isolation_prevents_real_home_writes(self): """The conftest autouse fixture sets HERMES_HOME; verify it's active.""" @@ -1187,17 +1185,14 @@ def test_conftest_isolation_prevents_real_home_writes(self): def test_get_hermes_home_fallback(self): """Without HERMES_HOME set, falls back to the active OS home.""" from tools.tirith_security import _get_hermes_home - env = {} - if os.name == "nt": - env["LOCALAPPDATA"] = r"C:\Users\test\AppData\Local" - with patch.dict(os.environ, env, clear=True): - expected = ( - os.path.join(env["LOCALAPPDATA"], "hermes") - if os.name == "nt" - else os.path.join(os.path.expanduser("~"), ".hermes") - ) + with patch.dict(os.environ, {}, clear=True): + # Remove HERMES_HOME entirely. With HOME also absent, expanduser + # falls back to the account database; compute expected under the + # same environment instead of after patch.dict restores HOME. + os.environ.pop("HERMES_HOME", None) + expected = os.path.join(os.path.expanduser("~"), ".hermes") result = _get_hermes_home() - assert os.path.normpath(result) == os.path.normpath(expected) + assert result == expected # --------------------------------------------------------------------------- @@ -1462,13 +1457,8 @@ class TestMkdtempOSErrorNoSpace: def test_mkdtemp_oserror_returns_no_space(self): from tools.tirith_security import _install_tirith - with patch( - "tools.tirith_security._detect_target", - return_value="x86_64-unknown-linux-gnu", - ), patch( - "tools.tirith_security.tempfile.mkdtemp", - side_effect=OSError(28, "No space left on device"), - ): + with patch("tools.tirith_security.tempfile.mkdtemp", + side_effect=OSError(28, "No space left on device")): result, reason = _install_tirith(log_failures=False) assert result is None assert reason == "no_space" @@ -1490,10 +1480,7 @@ def test_mkdtemp_oserror_propagates_to_ensure_installed(self): from tools.tirith_security import _resolve_tirith_path, _INSTALL_FAILED _tirith_mod._resolved_path = None - with patch( - "tools.tirith_security._detect_target", - return_value="x86_64-unknown-linux-gnu", - ), patch("tools.tirith_security.tempfile.mkdtemp", + with patch("tools.tirith_security.tempfile.mkdtemp", side_effect=OSError(28, "No space left on device")), \ patch("tools.tirith_security.shutil.which", return_value=None), \ diff --git a/tools/tirith_security.py b/tools/tirith_security.py index 99f2634b93a16..93509604131d1 100644 --- a/tools/tirith_security.py +++ b/tools/tirith_security.py @@ -385,12 +385,10 @@ def _extract_tirith_binary(tar: tarfile.TarFile, dest_dir: str, log) -> tuple[st def _install_tirith(*, log_failures: bool = True) -> tuple[str | None, str]: """Download and install tirith to $HERMES_HOME/bin/tirith. - Always verifies the SHA-256 checksum. When cosign is available, also - verifies release provenance; unavailable or operationally broken cosign - falls back to checksum verification. + Verifies provenance via cosign and SHA-256 checksum. Returns (installed_path, failure_reason). On success failure_reason is "". failure_reason is a short tag used by the disk marker to decide if the - failure is retryable. + failure is retryable (e.g. "cosign_missing" clears when cosign appears). """ log = logger.warning if log_failures else logger.debug @@ -423,32 +421,34 @@ def _install_tirith(*, log_failures: bool = True) -> tuple[str | None, str]: log("tirith download failed: %s", exc) return None, "download_failed" + # Cosign provenance verification — preferred but not mandatory. + # When cosign is available, we verify that the release was produced + # by the expected GitHub Actions workflow (full supply chain proof). + # Without cosign, SHA-256 checksum + HTTPS still provides integrity + # and transport-level authenticity. cosign_verified = False if shutil.which("cosign"): try: _download_file(f"{base_url}/checksums.txt.sig", sig_path) _download_file(f"{base_url}/checksums.txt.pem", cert_path) except Exception as exc: - logger.info( - "cosign artifacts unavailable (%s), proceeding with SHA-256 only", - exc, - ) + logger.info("cosign artifacts unavailable (%s), proceeding with SHA-256 only", exc) else: cosign_result = _verify_cosign(checksums_path, sig_path, cert_path) if cosign_result is True: cosign_verified = True elif cosign_result is False: + # Verification explicitly rejected — abort, the release + # may have been tampered with. log("tirith install aborted: cosign provenance verification failed") return None, "cosign_verification_failed" else: - logger.info( - "cosign execution failed, proceeding with SHA-256 only" - ) + # None = execution failure (timeout/OSError) — proceed + # with SHA-256 only since cosign itself is broken. + logger.info("cosign execution failed, proceeding with SHA-256 only") else: - logger.info( - "cosign not on PATH - installing tirith with SHA-256 verification only " - "(install cosign for full supply chain verification)" - ) + logger.info("cosign not on PATH — installing tirith with SHA-256 verification only " + "(install cosign for full supply chain verification)") if not _verify_checksum(archive_path, checksums_path, archive_name): return None, "checksum_failed" From b57d0bc64049f1c1c843747e55ebd9a8a77cfdbb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B3=AF=E5=B2=B8=E3=80=80=E4=BA=AE?= <1920071390@campus.ouj.ac.jp> Date: Sun, 12 Jul 2026 20:50:47 +0900 Subject: [PATCH 4/4] fix(security): harden backup archive validation --- hermes_cli/backup.py | 23 +++++++---- tests/hermes_cli/test_backup.py | 65 ++++++++++++++++++++++++++++++++ tests/tools/test_read_extract.py | 41 ++++++++++++++++++++ tools/read_extract.py | 4 +- 4 files changed, 125 insertions(+), 8 deletions(-) diff --git a/hermes_cli/backup.py b/hermes_cli/backup.py index 8caaf9618eda7..aafd39c2c38f5 100644 --- a/hermes_cli/backup.py +++ b/hermes_cli/backup.py @@ -11,6 +11,7 @@ import json import logging import os +import posixpath import shutil import sqlite3 import sys @@ -502,17 +503,24 @@ def _validate_backup_zip(zf: zipfile.ZipFile) -> tuple[bool, str]: return True, "" -def _validate_import_members(zf: zipfile.ZipFile) -> tuple[bool, str]: - infos = [info for info in zf.infolist() if not info.is_dir()] +def _validate_import_members(zf: zipfile.ZipFile, prefix: str) -> tuple[bool, str]: + infos = zf.infolist() if len(infos) > _MAX_IMPORT_MEMBERS: return False, f"zip contains too many files (maximum {_MAX_IMPORT_MEMBERS:,})" total_bytes = 0 seen_names: set[str] = set() + seen_normalized_names: set[str] = set() for info in infos: + if info.is_dir(): + continue if info.filename in seen_names: return False, f"zip contains a duplicate file name: {info.filename}" seen_names.add(info.filename) + normalized_name = _normalized_import_member_name(info.filename, prefix) + if normalized_name in seen_normalized_names: + return False, f"zip contains colliding file names: {info.filename}" + seen_normalized_names.add(normalized_name) if info.file_size > _MAX_IMPORT_MEMBER_BYTES: return False, f"zip member is too large: {info.filename}" total_bytes += info.file_size @@ -536,9 +544,10 @@ def _copy_zip_member( def _normalized_import_member_name(filename: str, prefix: str) -> str: - if prefix and filename.startswith(prefix): - return filename[len(prefix):] - return filename + normalized_filename = filename.replace("\\", "/") + if prefix and normalized_filename.startswith(prefix): + normalized_filename = normalized_filename[len(prefix):] + return posixpath.normpath(normalized_filename) def _is_allowed_external_target(target: Path, allowed_roots: List[Path]) -> bool: @@ -604,12 +613,12 @@ def run_import(args) -> None: print(f"Error: {reason}") sys.exit(1) - ok, reason = _validate_import_members(zf) + prefix = _detect_prefix(zf) + ok, reason = _validate_import_members(zf, prefix) if not ok: print(f"Error: {reason}") sys.exit(1) - prefix = _detect_prefix(zf) members = sorted( (info for info in zf.infolist() if not info.is_dir()), key=lambda info: _normalized_import_member_name( diff --git a/tests/hermes_cli/test_backup.py b/tests/hermes_cli/test_backup.py index 57389fe63f656..b91a63131a31c 100644 --- a/tests/hermes_cli/test_backup.py +++ b/tests/hermes_cli/test_backup.py @@ -1236,6 +1236,23 @@ def test_import_rejects_excessive_member_count(self, tmp_path, monkeypatch): with pytest.raises(SystemExit): backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) + def test_import_counts_directory_entries_toward_member_limit(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + zip_path = tmp_path / "too-many-directories.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("config.yaml", "model: test\n") + zf.writestr("one/", "") + zf.writestr("two/", "") + + import hermes_cli.backup as backup_mod + monkeypatch.setattr(backup_mod, "_MAX_IMPORT_MEMBERS", 2) + + with pytest.raises(SystemExit): + backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) + def test_import_rejects_excessive_archive_size(self, tmp_path, monkeypatch): hermes_home = tmp_path / ".hermes" hermes_home.mkdir() @@ -1265,6 +1282,37 @@ def test_import_rejects_excessive_compression_ratio(self, tmp_path, monkeypatch) with pytest.raises(SystemExit): backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) + def test_import_rejects_member_larger_than_limit(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + zip_path = tmp_path / "large-member.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("config.yaml", "model: test\n") + + import hermes_cli.backup as backup_mod + monkeypatch.setattr(backup_mod, "_MAX_IMPORT_MEMBER_BYTES", 1) + + with pytest.raises(SystemExit): + backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) + + def test_import_rejects_total_expansion_larger_than_limit(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + zip_path = tmp_path / "large-expansion.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("config.yaml", "model: test\n") + zf.writestr("sessions/one.json", "{}") + + import hermes_cli.backup as backup_mod + monkeypatch.setattr(backup_mod, "_MAX_IMPORT_TOTAL_BYTES", 1) + + with pytest.raises(SystemExit): + backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) + def test_import_rejects_duplicate_member_names(self, tmp_path, monkeypatch): hermes_home = tmp_path / ".hermes" hermes_home.mkdir() @@ -1280,6 +1328,23 @@ def test_import_rejects_duplicate_member_names(self, tmp_path, monkeypatch): with pytest.raises(SystemExit): backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) + def test_import_rejects_normalized_member_name_collision(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + (hermes_home / "config.yaml").write_text("original: true\n") + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + zip_path = tmp_path / "normalized-collision.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("config.yaml", "model: first\n") + zf.writestr(".hermes/config.yaml", "model: second\n") + + import hermes_cli.backup as backup_mod + with pytest.raises(SystemExit): + backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) + + assert (hermes_home / "config.yaml").read_text() == "original: true\n" + # --------------------------------------------------------------------------- # Profile restoration tests diff --git a/tests/tools/test_read_extract.py b/tests/tools/test_read_extract.py index fce198f6579a1..28ad65772ac36 100644 --- a/tests/tools/test_read_extract.py +++ b/tests/tools/test_read_extract.py @@ -188,6 +188,47 @@ def test_rejects_oversized_archive_member(self): finally: read_extract.MAX_OFFICE_MEMBER_BYTES = original + def test_rejects_excessive_archive_member_count(self): + p = os.path.join(self.tmp, "many-members.docx") + _write_docx(p, self._doc('Text')) + with zipfile.ZipFile(p, "a") as z: + z.writestr("metadata/", "") + + import tools.read_extract as read_extract + original = read_extract.MAX_OFFICE_MEMBER_COUNT + read_extract.MAX_OFFICE_MEMBER_COUNT = 2 + try: + with self.assertRaises(ExtractionError): + extract_document_text(p) + finally: + read_extract.MAX_OFFICE_MEMBER_COUNT = original + + def test_rejects_excessive_archive_expansion(self): + p = os.path.join(self.tmp, "large-total.docx") + _write_docx(p, self._doc('Text')) + + import tools.read_extract as read_extract + original = read_extract.MAX_OFFICE_TOTAL_BYTES + read_extract.MAX_OFFICE_TOTAL_BYTES = 1 + try: + with self.assertRaises(ExtractionError): + extract_document_text(p) + finally: + read_extract.MAX_OFFICE_TOTAL_BYTES = original + + def test_rejects_excessive_compressed_archive_size(self): + p = os.path.join(self.tmp, "large-compressed.docx") + _write_docx(p, self._doc('Text')) + + import tools.read_extract as read_extract + original = read_extract.MAX_XLSX_BYTES + read_extract.MAX_XLSX_BYTES = 1 + try: + with self.assertRaises(ExtractionError): + extract_document_text(p) + finally: + read_extract.MAX_XLSX_BYTES = original + # --------------------------------------------------------------------------- # Excel workbooks (.xlsx) — #10740 diff --git a/tools/read_extract.py b/tools/read_extract.py index 5c26cf3385c32..6e901dfc76ce6 100644 --- a/tools/read_extract.py +++ b/tools/read_extract.py @@ -115,12 +115,14 @@ def _validate_office_archive(path: str, zf: zipfile.ZipFile) -> None: except OSError as exc: raise ExtractionError(str(exc)) from exc - infos = [info for info in zf.infolist() if not info.is_dir()] + infos = zf.infolist() if len(infos) > MAX_OFFICE_MEMBER_COUNT: raise ExtractionError("Office document contains too many archive members") total_size = 0 for info in infos: + if info.is_dir(): + continue if info.file_size > MAX_OFFICE_MEMBER_BYTES: raise ExtractionError(f"Office document member is too large: {info.filename}") total_size += info.file_size