diff --git a/CHANGELOG.md b/CHANGELOG.md index 27be51b6..3a145d3c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,16 @@ ## [Unreleased] +## [v0.51.523] — 2026-06-19 — Release SH (cache trusted wiki page listings on browser reads) + +### Changed + +- **The LLM Wiki browse/read endpoints no longer re-walk the entire wiki tree on every request (#4375).** `/api/wiki/page` and `/api/wiki/browse` rebuilt the trusted page allowlist by walking up to 10k files (with a `stat` each) on every call (#3576 hardening). The walk result is now memoized in a short-TTL (5s) in-process cache keyed on the resolved wiki root with a section-dir change signature. The cache only memoizes which page names are candidates — the per-request read path still re-resolves each entry, re-asserts containment under the real wiki root (with dotfile exclusion), and gates the actual read behind the existing `O_NOFOLLOW` open + inode-identity check, so a cached entry can never bypass a containment check the fresh walk would apply. + +### Security + +- **Hardlinked wiki page files are now rejected from the allowlist and read revalidation (#4375).** `O_NOFOLLOW` plus inode-identity cannot distinguish a hardlink at a clean `*.md` page name from the real page, so a multi-link file at a listed name could carry an arbitrary inode (including one outside the wiki) through the read check. Any page file with `st_nlink > 1` is now excluded during both the allowlist walk and the read-path revalidation. Thanks @rodboev. + ## [v0.51.522] — 2026-06-19 — Release SG (live model lookups never forward another provider's key) ### Security diff --git a/api/routes.py b/api/routes.py index 8db3c059..2cbdfd8d 100644 --- a/api/routes.py +++ b/api/routes.py @@ -17,6 +17,7 @@ import platform import shlex import shutil import sqlite3 +import stat as _stat import subprocess import sys import threading @@ -5804,6 +5805,9 @@ _LLM_WIKI_MAX_PAGE_BYTES = 2 * 1024 * 1024 _LLM_WIKI_FORBIDDEN_ROOTS = frozenset( str(Path(p).expanduser().resolve()) for p in ("/", "/etc", "/usr", "/var", "/opt", "/sys", "/proc") ) +_WIKI_ALLOWLIST_TTL = 5.0 # seconds +_wiki_allowlist_cache: dict[str, dict[str, object]] = {} +_wiki_allowlist_cache_lock = threading.Lock() def _llm_wiki_resolve_path() -> tuple[Path, str, bool]: @@ -5856,7 +5860,28 @@ def _llm_wiki_count_files(root: Path) -> int: return count -def _llm_wiki_page_files(wiki_path: Path) -> list[Path]: +def _llm_wiki_page_files_cache_signature(wiki_path: Path) -> tuple: + """Return a change-detection signature over the configured wiki sections. + + Reads only the section-level directories (not every page), using + st_mtime_ns, st_dev, and st_ino so that quick section replacements and + mtime-ns-resolution changes are both caught. Missing or inaccessible + section dirs are represented as ``(section, None)`` so their appearance + or disappearance also invalidates the cache. + """ + sig = [] + for section in _LLM_WIKI_PAGE_DIRS: + section_dir = wiki_path / section + try: + st = section_dir.lstat() + except OSError: + sig.append((section, None)) + continue + sig.append((section, st.st_dev, st.st_ino, st.st_mtime_ns)) + return tuple(sig) + + +def _llm_wiki_page_files_uncached(wiki_path: Path) -> list[Path]: pages: list[Path] = [] # Defense in depth: refuse forbidden system roots, and resolve the wiki root # ONCE as the single trust base for all containment checks below. @@ -5892,6 +5917,16 @@ def _llm_wiki_page_files(wiki_path: Path) -> list[Path]: rel = item.relative_to(section) if not item.is_file() or not _is_clean_relpath(rel): continue + # Reject multi-link (hardlinked) page files. A hardlink at a + # clean *.md name can point at an arbitrary inode (incl. one + # outside the wiki); O_NOFOLLOW + inode-identity at read time + # cannot tell such a hardlink apart from the real page, so + # exclude any file with st_nlink > 1 from the allowlist. (#4375) + try: + if item.lstat().st_nlink > 1: + continue + except OSError: + continue # Resolve the real target and require it to live under BOTH the # real wiki root and the real section, with no dot-prefixed # segment on the resolved-relative path. This closes symlink @@ -5909,7 +5944,136 @@ def _llm_wiki_page_files(wiki_path: Path) -> list[Path]: return pages -def _llm_wiki_last_writer(wiki_path: Path, page_files: list[Path]) -> str: +def _llm_wiki_page_files(wiki_path: Path) -> list[Path]: + """Return all allowlisted wiki page paths under *wiki_path*. + + Trust boundary: the allowlist uses ``(st_dev, st_ino)`` inode identity. + A hardlink created at a listed page name *before* this snapshot is taken + would carry that page's identity through to the caller's fstat check. + This is only exploitable with write access to the wiki directory, which + is outside the realistic threat model (the wiki directory is + operator-controlled). Defense-in-depth via an ``openat``-chain with + no-follow directory fds would close the gap but is deferred — the inode + match already defeats path-swap and symlink races at open time. + """ + try: + wiki_root = wiki_path.resolve() + except OSError: + wiki_root = wiki_path + + key = str(wiki_root) + sig = _llm_wiki_page_files_cache_signature(wiki_root) + now = time.monotonic() + + with _wiki_allowlist_cache_lock: + cached = _wiki_allowlist_cache.get(key) + if ( + cached is not None + and cached.get("signature") == sig + and now < cached.get("expires_at", 0) + ): + return list(cached["files"]) + + pages = _llm_wiki_page_files_uncached(wiki_root) + + with _wiki_allowlist_cache_lock: + _wiki_allowlist_cache[key] = { + "signature": sig, + "expires_at": now + _WIKI_ALLOWLIST_TTL, + "files": tuple(pages), + } + return list(pages) + + +def _llm_wiki_clear_page_files_cache() -> None: + """Clear the allowlist cache; intended for tests only.""" + with _wiki_allowlist_cache_lock: + _wiki_allowlist_cache.clear() + + +def _llm_wiki_allowlisted_entries(wiki_path: Path) -> dict[str, tuple[Path, tuple[int, int]]]: + """Return listed relpaths mapped to resolved targets plus stable identity.""" + try: + wiki_real = wiki_path.resolve() + except OSError: + return {} + + def _wiki_read_relpath_is_clean(rel: Path) -> bool: + rel_text = rel.as_posix() + return bool(rel.parts) and "\\" not in rel_text and not any(part.startswith(".") for part in rel.parts) + + entries: dict[str, tuple[Path, tuple[int, int]]] = {} + try: + for listed_path in _llm_wiki_page_files(wiki_real): + try: + rel_listed = listed_path.relative_to(wiki_real) + if not _wiki_read_relpath_is_clean(rel_listed): + continue + section_real = (wiki_real / rel_listed.parts[0]).resolve() + section_real.relative_to(wiki_real) + resolved_target = listed_path.resolve() + resolved_target.relative_to(section_real) + if not resolved_target.is_file(): + continue + # Reject hardlinked targets (st_nlink > 1): a multi-link file at + # a clean page name can carry an arbitrary inode through the + # O_NOFOLLOW + identity read check. Defense in depth alongside + # the same rejection in the allowlist walk. (#4375) + if resolved_target.stat().st_nlink > 1: + continue + rel_resolved = resolved_target.relative_to(wiki_real) + if not _wiki_read_relpath_is_clean(rel_resolved): + continue + st0 = resolved_target.stat() + if listed_path.resolve() != resolved_target: + continue + entries[rel_listed.as_posix()] = (resolved_target, (st0.st_dev, st0.st_ino)) + except (OSError, ValueError): + continue + except Exception: + return {} + return entries + + +def _llm_wiki_status_file_entry_stat(wiki_path: Path, path: Path) -> os.stat_result | None: + """Return the current top-level status-file entry metadata when it is safe to trust.""" + try: + wiki_root = wiki_path.resolve() + except OSError: + wiki_root = wiki_path + try: + path.resolve().relative_to(wiki_root) + st_entry = path.lstat() + except (OSError, ValueError): + return None + if not _stat.S_ISREG(st_entry.st_mode): + return None + return st_entry + + +def _llm_wiki_verified_status_file_stat(wiki_path: Path, path: Path) -> os.stat_result | None: + """Return identity-checked metadata for a top-level wiki status file.""" + st_entry = _llm_wiki_status_file_entry_stat(wiki_path, path) + if st_entry is None: + return None + fd = None + try: + fd = os.open(str(path), os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) + st_open = os.fstat(fd) + if (st_open.st_dev, st_open.st_ino) != (st_entry.st_dev, st_entry.st_ino): + return None + return st_open + except OSError: + return None + finally: + if fd is not None: + os.close(fd) + + +def _llm_wiki_last_writer( + wiki_path: Path, + page_files: list[Path] | list[tuple[Path, tuple[int, int]]], +) -> str: """Best-effort last-writer detection for the LLM Wiki status card. Closes the gap left by the original panel (commit 2684d6fa, Issue #1257): @@ -5944,52 +6108,78 @@ def _llm_wiki_last_writer(wiki_path: Path, page_files: list[Path]) -> str: # Priority 1: most recent page frontmatter (resolved-path must stay in-wiki) latest_page: Path | None = None + latest_identity: tuple[int, int] | None = None latest_mtime = -1.0 - for candidate in page_files: + for item in page_files: + candidate, identity = item if isinstance(item, tuple) else (item, None) if not _within_wiki(candidate): continue # skip symlinks resolving outside the wiki try: - mtime = candidate.stat().st_mtime + st_candidate = candidate.stat() except Exception: continue + if identity is not None and (st_candidate.st_dev, st_candidate.st_ino) != identity: + continue + mtime = st_candidate.st_mtime if mtime > latest_mtime: latest_mtime = mtime latest_page = candidate + latest_identity = identity if latest_page is not None: try: - with open(latest_page, encoding="utf-8", errors="replace") as fh: - first = fh.readline() - if first.strip() == "---": - # Read only the frontmatter block, bounded to a small line cap. - for _ in range(200): - line = fh.readline() - if line == "" or line.strip() == "---": - break # EOF or end of frontmatter — never touch the body - stripped = line.strip() - lower = stripped.lower() - for key in ("updated_by", "writer", "author"): - if lower.startswith(f"{key}:"): - value = stripped.split(":", 1)[1].strip() - if value: - return value + fd = os.open(str(latest_page), os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) + try: + if latest_identity is not None: + st_open = os.fstat(fd) + if (st_open.st_dev, st_open.st_ino) != latest_identity: + raise FileNotFoundError("wiki page changed after allowlist snapshot") + with os.fdopen(fd, encoding="utf-8", errors="replace") as fh: + fd = None + first = fh.readline() + if first.strip() == "---": + # Read only the frontmatter block, bounded to a small line cap. + for _ in range(200): + line = fh.readline() + if line == "" or line.strip() == "---": + break # EOF or end of frontmatter — never touch the body + stripped = line.strip() + lower = stripped.lower() + for key in ("updated_by", "writer", "author"): + if lower.startswith(f"{key}:"): + value = stripped.split(":", 1)[1].strip() + if value: + return value + finally: + if fd is not None: + os.close(fd) except Exception: pass # Priority 2: log.md last entry action verb (heading lines only, bounded) log_path = wiki_path / "log.md" - if _within_wiki(log_path) and log_path.is_file(): + log_entry = _llm_wiki_status_file_entry_stat(wiki_path, log_path) + if log_entry is not None: try: - with open(log_path, encoding="utf-8", errors="replace") as fh: - for _ in range(5000): # cap the heading scan - line = fh.readline() - if line == "": - break - stripped = line.strip() - if not stripped.startswith("## [") or "|" not in stripped: - continue - tail = stripped.split("]", 1)[1].strip() if "]" in stripped else "" - action = tail.split()[0] if tail else "update" - return f"ai-agent ({action})" + fd = os.open(str(log_path), os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) + try: + st_open = os.fstat(fd) + if (st_open.st_dev, st_open.st_ino) != (log_entry.st_dev, log_entry.st_ino): + raise FileNotFoundError("wiki log changed before open") + with os.fdopen(fd, encoding="utf-8", errors="replace") as fh: + fd = None + for _ in range(5000): # cap the heading scan + line = fh.readline() + if line == "": + break + stripped = line.strip() + if not stripped.startswith("## [") or "|" not in stripped: + continue + tail = stripped.split("]", 1)[1].strip() if "]" in stripped else "" + action = tail.split()[0] if tail else "update" + return f"ai-agent ({action})" + finally: + if fd is not None: + os.close(fd) except Exception: pass @@ -6022,15 +6212,27 @@ def _build_llm_wiki_status() -> dict: base["status"] = "not_directory" return base - page_files = _llm_wiki_page_files(wiki_path) - status_files = [p for p in (wiki_path / "SCHEMA.md", wiki_path / "index.md", wiki_path / "log.md") if p.exists() and p.is_file()] - status_files.extend(page_files) - latest = None - for item in status_files: + allowlisted_entries = _llm_wiki_allowlisted_entries(wiki_path) + verified_page_entries: list[tuple[Path, tuple[int, int], os.stat_result]] = [] + for target, identity in allowlisted_entries.values(): try: - mtime = item.stat().st_mtime + st = target.stat() except Exception: continue + if (st.st_dev, st.st_ino) != identity: + continue + verified_page_entries.append((target, identity, st)) + page_entries = [(target, identity) for target, identity, _ in verified_page_entries] + page_files = [target for target, _, _ in verified_page_entries] + status_files: list[tuple[Path, float]] = [] + for path in (wiki_path / "SCHEMA.md", wiki_path / "index.md", wiki_path / "log.md"): + st_status = _llm_wiki_verified_status_file_stat(wiki_path, path) + if st_status is None: + continue + status_files.append((path, st_status.st_mtime)) + status_files.extend((target, st.st_mtime) for target, _, st in verified_page_entries) + latest = None + for _, mtime in status_files: latest = mtime if latest is None else max(latest, mtime) base.update({ @@ -6038,10 +6240,10 @@ def _build_llm_wiki_status() -> dict: "enabled": True, "status": "ready" if page_files else "empty", "entry_count": len(page_files), - "page_count": len(page_files), + "page_count": len(page_entries), "raw_source_count": _llm_wiki_count_files(wiki_path / "raw"), "last_updated": _llm_wiki_safe_iso(latest), - "last_writer": _llm_wiki_last_writer(wiki_path, page_files), + "last_writer": _llm_wiki_last_writer(wiki_path, page_entries), }) return base except Exception as exc: @@ -7265,34 +7467,41 @@ def handle_get(handler, parsed) -> bool: wiki_root, _, _ = _llm_wiki_resolve_path() if not wiki_root or not os.path.isdir(wiki_root): return bad(handler, "Wiki not configured or directory not found", status=404) - page_paths = _llm_wiki_page_files(wiki_root) + allowlisted_entries = _llm_wiki_allowlisted_entries(Path(wiki_root)) pages = [] - for fp in sorted(page_paths, key=lambda p: str(p).lower()): - try: - rel = fp.relative_to(wiki_root) - except ValueError: - continue + for rel_path, (fp, identity) in sorted(allowlisted_entries.items(), key=lambda item: item[0].lower()): try: st = fp.stat() except OSError: continue - pages.append({"name": fp.name, "path": str(rel).replace("\\", "/"), "size": st.st_size, "mtime": int(st.st_mtime)}) + if (st.st_dev, st.st_ino) != identity: + continue + pages.append({"name": Path(rel_path).name, "path": rel_path, "size": st.st_size, "mtime": int(st.st_mtime)}) return j(handler, {"pages": pages}) if parsed.path == "/api/wiki/page": wiki_root, _, _ = _llm_wiki_resolve_path() page_path = parse_qs(parsed.query or "").get("path", [""])[0] if not wiki_root or not page_path: return bad(handler, "Wiki not configured or path not provided", status=400) + if "\\" in page_path: + return bad(handler, "Invalid path", status=400) # Reject a real `..` path SEGMENT (or absolute path), not the bare # substring — a legitimate listed filename like `v1..v2.md` contains # ".." without being traversal. Containment + the resolved-allowlist # membership check below are the actual security boundary. - _page_parts = page_path.replace("\\", "/").split("/") + requested_key = page_path.replace("\\", "/") + _page_parts = requested_key.split("/") if os.path.isabs(page_path) or any(part == ".." for part in _page_parts): return bad(handler, "Invalid path", status=400) + if any(part in ("", ".") for part in _page_parts): + return bad(handler, "Invalid path", status=400) full_path = Path(os.path.join(wiki_root, page_path)) if not _skill_path_within(Path(wiki_root), full_path): return bad(handler, "Invalid path", status=400) + try: + wiki_real = Path(wiki_root).resolve() + except OSError: + return bad(handler, "Page not found", status=404) # Only serve files the browse/list path would surface (same allowlist: # *.md under the wiki page-dirs, no dotfiles, forbidden-roots guard). # Without this the read endpoint could return ANY file inside the wiki @@ -7303,22 +7512,16 @@ def handle_get(handler, parsed) -> bool: # allowlist check (TOCTOU write-race, Codex finding) — a pathname re-open # alone can't, since O_NOFOLLOW only guards the final component, not a # swapped parent directory. - allowed_identity: dict[Path, tuple] = {} - try: - for _p in _llm_wiki_page_files(Path(wiki_root)): - try: - rp = _p.resolve() - st0 = rp.stat() - allowed_identity[rp] = (st0.st_dev, st0.st_ino) - except OSError: - continue - except Exception: - allowed_identity = {} + allowed_identity = _llm_wiki_allowlisted_entries(wiki_real) try: resolved_target = full_path.resolve() except OSError: return bad(handler, "Page not found", status=404) - if resolved_target not in allowed_identity: + requested_entry = allowed_identity.get(requested_key) + if requested_entry is None: + return bad(handler, "Page not found", status=404) + allowlisted_target, allowlisted_identity = requested_entry + if resolved_target != allowlisted_target: return bad(handler, "Page not found", status=404) # Read the ALREADY-RESOLVED, allowlisted real path with O_NOFOLLOW so a # symlink swapped in for the final component between the allowlist check @@ -7331,7 +7534,7 @@ def handle_get(handler, parsed) -> bool: fd = os.open(str(resolved_target), os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) try: st_open = os.fstat(fd) - if (st_open.st_dev, st_open.st_ino) != allowed_identity[resolved_target]: + if (st_open.st_dev, st_open.st_ino) != allowlisted_identity: return bad(handler, "Page not found", status=404) raw = os.read(fd, _LLM_WIKI_MAX_PAGE_BYTES + 1) finally: diff --git a/tests/test_issue1257_llm_wiki_status.py b/tests/test_issue1257_llm_wiki_status.py index 22b23250..4540db90 100644 --- a/tests/test_issue1257_llm_wiki_status.py +++ b/tests/test_issue1257_llm_wiki_status.py @@ -112,6 +112,180 @@ def test_last_writer_reads_frontmatter(tmp_path): assert routes._llm_wiki_last_writer(wiki, pages) == "alice" +def test_llm_wiki_status_rechecks_cached_page_targets(monkeypatch, tmp_path): + import os as _os + import api.routes as routes + + wiki = tmp_path / "wiki" + page = _write(wiki / "concepts" / "sub" / "real.md", "---\nauthor: public\n---\nbody\n") + _write(wiki / ".env", "---\nauthor: hidden-author\n---\nPRIVATE=1\n") + + routes._llm_wiki_clear_page_files_cache() + monkeypatch.setenv("WIKI_PATH", str(wiki)) + monkeypatch.setattr(routes, "_WIKI_ALLOWLIST_TTL", 60.0) + + assert routes._llm_wiki_page_files(wiki) == [page] + + page.unlink() + try: + page.symlink_to(_os.path.join("..", "..", ".env")) + except (OSError, NotImplementedError): + import pytest + pytest.skip("symlinks not supported on this platform") + + status = routes._build_llm_wiki_status() + + assert status["page_count"] == 0 + assert status["last_writer"] == "ai-agent" + assert "hidden-author" not in repr(status) + + +def test_llm_wiki_status_drops_cached_entry_replaced_by_directory(monkeypatch, tmp_path): + import api.routes as routes + + wiki = tmp_path / "wiki" + page = _write(wiki / "concepts" / "sub" / "real.md", "---\nauthor: public\n---\nbody\n") + + routes._llm_wiki_clear_page_files_cache() + monkeypatch.setenv("WIKI_PATH", str(wiki)) + monkeypatch.setattr(routes, "_WIKI_ALLOWLIST_TTL", 60.0) + + assert routes._llm_wiki_page_files(wiki) == [page] + + page.unlink() + page.mkdir() + + status = routes._build_llm_wiki_status() + + assert status["page_count"] == 0 + assert status["last_writer"] == "ai-agent" + + +def test_llm_wiki_status_last_writer_rechecks_identity(monkeypatch, tmp_path): + import os as _os + import api.routes as routes + + wiki = tmp_path / "wiki" + page = _write(wiki / "concepts" / "real.md", "---\nauthor: public\n---\nbody\n") + _write(wiki / ".env", "---\nauthor: hidden-author\n---\nPRIVATE=1\n") + + routes._llm_wiki_clear_page_files_cache() + monkeypatch.setenv("WIKI_PATH", str(wiki)) + + original = routes._llm_wiki_allowlisted_entries + + def swapped(root): + entries = original(root) + page.unlink() + page.symlink_to(_os.path.join("..", ".env")) + return entries + + monkeypatch.setattr(routes, "_llm_wiki_allowlisted_entries", swapped) + + status = routes._build_llm_wiki_status() + + assert status["page_count"] == 0 + assert status["last_updated"] is None + assert status["last_writer"] == "ai-agent" + assert "hidden-author" not in repr(status) + + +def test_llm_wiki_status_log_heading_rechecks_identity(monkeypatch, tmp_path): + import os as _os + import api.routes as routes + + wiki = tmp_path / "wiki" + _write(wiki / "log.md", "## [2026-06-19] update | safe\n",) + _write(wiki / ".env", "## [2026-06-19] leak | hidden\n") + + monkeypatch.setenv("WIKI_PATH", str(wiki)) + + original_open = routes.os.open + swapped = {"done": False} + + def fake_open(path, flags, mode=0o777): + if not swapped["done"] and Path(path) == wiki / "log.md": + swapped["done"] = True + (wiki / "log.md").unlink() + (wiki / "log.md").symlink_to(_os.path.join(".env")) + return original_open(path, flags, mode) + + monkeypatch.setattr(routes.os, "open", fake_open) + + status = routes._build_llm_wiki_status() + + assert status["last_writer"] == "ai-agent" + + +def test_llm_wiki_status_log_heading_rechecks_preopen_symlink_swap(monkeypatch, tmp_path): + import api.routes as routes + + wiki = tmp_path / "wiki" + _write(wiki / "log.md", "## [2026-06-19] update | safe\n") + _write(wiki / ".env", "## [2026-06-19] leak | hidden\n") + + try: + link = wiki / "probe" + link.symlink_to(".env") + link.unlink() + except (OSError, NotImplementedError): + import pytest + pytest.skip("symlinks not supported on this platform") + + monkeypatch.setenv("WIKI_PATH", str(wiki)) + + original_lstat = Path.lstat + swapped = {"done": False} + + def fake_lstat(self): + if not swapped["done"] and self == wiki / "log.md": + swapped["done"] = True + (wiki / "log.md").unlink() + (wiki / "log.md").symlink_to(".env") + return original_lstat(self) + + monkeypatch.setattr(Path, "lstat", fake_lstat) + + status = routes._build_llm_wiki_status() + + assert status["last_writer"] == "ai-agent" + + +def test_llm_wiki_status_last_updated_rechecks_status_file_identity(monkeypatch, tmp_path): + import api.routes as routes + + wiki = tmp_path / "wiki" + log_path = _write(wiki / "log.md", "## [2026-06-19] update | safe\n") + hidden = _write(wiki / ".env", "PRIVATE=1\n") + + try: + log_path.unlink() + log_path.symlink_to(hidden.name) + log_path.unlink() + log_path.write_text("## [2026-06-19] update | safe\n", encoding="utf-8") + except (OSError, NotImplementedError): + import pytest + pytest.skip("symlinks not supported on this platform") + + monkeypatch.setenv("WIKI_PATH", str(wiki)) + + original_stat = Path.stat + swapped = {"done": False} + + def fake_stat(self, *args, **kwargs): + if not swapped["done"] and self == log_path: + swapped["done"] = True + log_path.unlink() + log_path.symlink_to(hidden.name) + return original_stat(self, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", fake_stat) + + status = routes._build_llm_wiki_status() + + assert status["last_updated"] is None + + def test_last_writer_rejects_symlink_outside_wiki(tmp_path): """#3455 review (Codex): a symlinked .md page resolving OUTSIDE the wiki must not be read — its frontmatter must never leak into the status card.""" diff --git a/tests/test_issue2941_wiki_browse.py b/tests/test_issue2941_wiki_browse.py index e6cb48f1..1ba47b19 100644 --- a/tests/test_issue2941_wiki_browse.py +++ b/tests/test_issue2941_wiki_browse.py @@ -9,6 +9,7 @@ Verifies that: from __future__ import annotations import json +import os import re from pathlib import Path from urllib.parse import urlparse @@ -82,7 +83,14 @@ def test_wiki_browse_skips_pages_that_disappear_during_listing(monkeypatch, tmp_ missing = wiki_root / "gone.md" monkeypatch.setattr(routes, "_llm_wiki_resolve_path", lambda: (wiki_root, None, None)) - monkeypatch.setattr(routes, "_llm_wiki_page_files", lambda root: [missing, ok]) + monkeypatch.setattr( + routes, + "_llm_wiki_allowlisted_entries", + lambda root: { + "gone.md": (missing, (0, 0)), + "ok.md": (ok, (ok.stat().st_dev, ok.stat().st_ino)), + }, + ) handler = _FakeHandler() routes.handle_get(handler, urlparse("http://example.com/api/wiki/browse")) @@ -105,31 +113,10 @@ def test_wiki_page_vanished_between_check_and_read_returns_404_not_500(monkeypat wiki_root = tmp_path / "wiki" wiki_root.mkdir() + missing = (wiki_root / "gone.md").resolve() monkeypatch.setattr(routes, "_llm_wiki_resolve_path", lambda: (wiki_root, None, None)) - - # Path resolves inside the root and passes the containment guard, but the - # read itself raises FileNotFoundError (simulating a vanished/racing file). - class _GonePath(type(wiki_root)): - def is_file(self): # noqa: D401 - test stub - return True - - def read_text(self, *a, **k): - raise FileNotFoundError("vanished between list and read") - - real_path_join = routes.os.path.join - - monkeypatch.setattr(routes, "_skill_path_within", lambda root, p: True) - - orig_Path = routes.Path - - def _fake_Path(arg): - # Only wrap the wiki page target; leave the root resolution alone. - if isinstance(arg, str) and arg == real_path_join(str(wiki_root), "gone.md"): - return _GonePath(arg) - return orig_Path(arg) - - monkeypatch.setattr(routes, "Path", _fake_Path) + monkeypatch.setattr(routes, "_llm_wiki_allowlisted_entries", lambda root: {"gone.md": (missing, (0, 0))}) handler = _FakeHandler() routes.handle_get(handler, urlparse("http://example.com/api/wiki/page?path=gone.md")) @@ -232,6 +219,281 @@ def test_wiki_symlink_to_hidden_same_section_target_blocked(monkeypatch, tmp_pat assert b"hidden_marker_q" not in h_page.body, "hidden secret leaked via symlink" +def test_wiki_page_cached_nested_entry_rechecks_resolved_containment(monkeypatch, tmp_path): + import os as _os + from api import routes + + wiki_root = tmp_path / "wiki" + section = wiki_root / "concepts" + nested = section / "sub" + nested.mkdir(parents=True) + page = nested / "real.md" + page.write_text("# real\n", encoding="utf-8") + secret = wiki_root / ".env" + secret.write_text("DONOTLEAK=stale_cache_marker\n", encoding="utf-8") + + try: + page.unlink() + page.symlink_to(_os.path.join("..", "..", ".env")) + except (OSError, NotImplementedError): + import pytest + pytest.skip("symlinks not supported on this platform") + + routes._llm_wiki_clear_page_files_cache() + monkeypatch.setattr(routes, "_WIKI_ALLOWLIST_TTL", 60.0) + monkeypatch.setattr(routes, "_llm_wiki_resolve_path", lambda: (wiki_root, None, None)) + + # Prime the cached name list while the nested page is still valid, then + # swap only the nested entry so the top-level section signature stays stale. + page.unlink() + page.write_text("# real\n", encoding="utf-8") + assert routes._llm_wiki_page_files(wiki_root) == [page] + sig_before = routes._llm_wiki_page_files_cache_signature(wiki_root.resolve()) + + page.unlink() + page.symlink_to(_os.path.join("..", "..", ".env")) + sig_after = routes._llm_wiki_page_files_cache_signature(wiki_root.resolve()) + assert sig_after == sig_before + + handler = _FakeHandler() + routes.handle_get(handler, urlparse("http://example.com/api/wiki/page?path=concepts/sub/real.md")) + + assert handler.status == 404, f"stale cached nested symlink swap must 404, got {handler.status}" + assert b"stale_cache_marker" not in handler.body, "stale cached nested symlink leaked hidden content" + + +def test_wiki_page_cached_entry_rejects_hardlink_swap_to_outside(monkeypatch, tmp_path): + """A hardlink swap to an outside file within the TTL must not leak (#4375). + + O_NOFOLLOW + inode-identity cannot distinguish a hardlink at a clean page + name from the real page, so multi-link (st_nlink > 1) page files are + rejected by both the allowlist walk and the read-path revalidation. Prime + the cache with a real page, replace it with a hardlink to an outside secret + within the TTL, and assert the read 404s without leaking the secret. + """ + import os as _os + from api import routes + + wiki_root = tmp_path / "wiki" + section = wiki_root / "concepts" + section.mkdir(parents=True) + page = section / "page.md" + page.write_text("# real\n", encoding="utf-8") + outside = tmp_path / "outside_secret" + outside.write_text("DONOTLEAK=hardlink_cache_marker\n", encoding="utf-8") + + routes._llm_wiki_clear_page_files_cache() + monkeypatch.setattr(routes, "_WIKI_ALLOWLIST_TTL", 60.0) + monkeypatch.setattr(routes, "_llm_wiki_resolve_path", lambda: (wiki_root, None, None)) + + # Prime the cached name list while the page is a genuine single-link file. + assert routes._llm_wiki_page_files(wiki_root) == [page] + + # Swap the listed page name to a hardlink pointing at the outside secret, + # within the TTL window (cache still warm for the page NAME). + page.unlink() + try: + _os.link(outside, page) + except (OSError, NotImplementedError, AttributeError): + import pytest + pytest.skip("hardlinks not supported on this platform") + + handler = _FakeHandler() + routes.handle_get(handler, urlparse("http://example.com/api/wiki/page?path=concepts/page.md")) + + assert handler.status == 404, f"cached page hardlinked to outside file must 404, got {handler.status}" + assert b"hardlink_cache_marker" not in handler.body, "hardlink swap leaked outside content" + + +def test_wiki_browse_cached_entry_rechecks_resolved_containment(monkeypatch, tmp_path): + import os as _os + from api import routes + + wiki_root = tmp_path / "wiki" + section = wiki_root / "concepts" + nested = section / "sub" + nested.mkdir(parents=True) + page = nested / "real.md" + page.write_text("# real\n", encoding="utf-8") + env_file = wiki_root / ".env" + env_file.write_text("DONOTLEAK=browse_cache_marker\n", encoding="utf-8") + + routes._llm_wiki_clear_page_files_cache() + monkeypatch.setattr(routes, "_WIKI_ALLOWLIST_TTL", 60.0) + monkeypatch.setattr(routes, "_llm_wiki_resolve_path", lambda: (wiki_root, None, None)) + + assert routes._llm_wiki_page_files(wiki_root) == [page] + + page.unlink() + try: + page.symlink_to(_os.path.join("..", "..", ".env")) + except (OSError, NotImplementedError): + import pytest + pytest.skip("symlinks not supported on this platform") + + handler = _FakeHandler() + routes.handle_get(handler, urlparse("http://example.com/api/wiki/browse")) + + assert handler.status == 200 + assert handler.get_json()["pages"] == [] + + +def test_wiki_browse_drops_cached_entry_replaced_by_directory(monkeypatch, tmp_path): + from api import routes + + wiki_root = tmp_path / "wiki" + section = wiki_root / "concepts" + nested = section / "sub" + nested.mkdir(parents=True) + page = nested / "real.md" + page.write_text("# real\n", encoding="utf-8") + + routes._llm_wiki_clear_page_files_cache() + monkeypatch.setattr(routes, "_WIKI_ALLOWLIST_TTL", 60.0) + monkeypatch.setattr(routes, "_llm_wiki_resolve_path", lambda: (wiki_root, None, None)) + + assert routes._llm_wiki_page_files(wiki_root) == [page] + + page.unlink() + page.mkdir() + + handler = _FakeHandler() + routes.handle_get(handler, urlparse("http://example.com/api/wiki/browse")) + + assert handler.status == 200 + assert handler.get_json()["pages"] == [] + + +def test_wiki_browse_rechecks_identity_after_allowlist_snapshot(monkeypatch, tmp_path): + import os as _os + from api import routes + + wiki_root = tmp_path / "wiki" + page = wiki_root / "concepts" / "real.md" + page.parent.mkdir(parents=True) + page.write_text("# real\n", encoding="utf-8") + (wiki_root / ".env").write_text("DONOTLEAK=post_snapshot_marker\n", encoding="utf-8") + + routes._llm_wiki_clear_page_files_cache() + monkeypatch.setattr(routes, "_llm_wiki_resolve_path", lambda: (wiki_root, None, None)) + + original = routes._llm_wiki_allowlisted_entries + + def swapped(root): + entries = original(root) + page.unlink() + page.symlink_to(_os.path.join("..", ".env")) + return entries + + monkeypatch.setattr(routes, "_llm_wiki_allowlisted_entries", swapped) + + handler = _FakeHandler() + routes.handle_get(handler, urlparse("http://example.com/api/wiki/browse")) + + assert handler.status == 200 + assert handler.get_json()["pages"] == [] + + +def test_wiki_page_cached_entry_cannot_jump_sections(monkeypatch, tmp_path): + import os as _os + from api import routes + + wiki_root = tmp_path / "wiki" + page = wiki_root / "concepts" / "sub" / "real.md" + page.parent.mkdir(parents=True) + page.write_text("# real\n", encoding="utf-8") + external_doc = wiki_root / "drafts" / "page.md" + external_doc.parent.mkdir(parents=True) + external_doc.write_text("cross_section_marker", encoding="utf-8") + + try: + page.unlink() + page.symlink_to(_os.path.join("..", "..", "drafts", "page.md")) + except (OSError, NotImplementedError): + import pytest + pytest.skip("symlinks not supported on this platform") + + routes._llm_wiki_clear_page_files_cache() + monkeypatch.setattr(routes, "_WIKI_ALLOWLIST_TTL", 60.0) + monkeypatch.setattr(routes, "_llm_wiki_resolve_path", lambda: (wiki_root, None, None)) + + page.unlink() + page.write_text("# real\n", encoding="utf-8") + assert routes._llm_wiki_page_files(wiki_root) == [page] + + page.unlink() + page.symlink_to(_os.path.join("..", "..", "drafts", "page.md")) + + handler = _FakeHandler() + routes.handle_get(handler, urlparse("http://example.com/api/wiki/page?path=concepts/sub/real.md")) + + assert handler.status == 404, f"stale cached cross-section swap must 404, got {handler.status}" + assert b"cross_section_marker" not in handler.body, "stale cached page leaked another section's content" + + +def test_wiki_page_cached_entry_cannot_jump_to_other_allowlisted_page(monkeypatch, tmp_path): + import os as _os + from api import routes + + wiki_root = tmp_path / "wiki" + page = wiki_root / "concepts" / "sub" / "real.md" + page.parent.mkdir(parents=True) + page.write_text("# real\n", encoding="utf-8") + sibling = wiki_root / "entities" / "page.md" + sibling.parent.mkdir(parents=True) + sibling.write_text("allowlisted_elsewhere_marker", encoding="utf-8") + + try: + page.unlink() + page.symlink_to(_os.path.join("..", "..", "entities", "page.md")) + except (OSError, NotImplementedError): + import pytest + pytest.skip("symlinks not supported on this platform") + + routes._llm_wiki_clear_page_files_cache() + monkeypatch.setattr(routes, "_WIKI_ALLOWLIST_TTL", 60.0) + monkeypatch.setattr(routes, "_llm_wiki_resolve_path", lambda: (wiki_root, None, None)) + + page.unlink() + page.write_text("# real\n", encoding="utf-8") + assert set(routes._llm_wiki_page_files(wiki_root)) == {page, sibling} + + page.unlink() + page.symlink_to(_os.path.join("..", "..", "entities", "page.md")) + + handler = _FakeHandler() + routes.handle_get(handler, urlparse("http://example.com/api/wiki/page?path=concepts/sub/real.md")) + + assert handler.status == 404, f"stale cached swap to another allowlisted page must 404, got {handler.status}" + assert b"allowlisted_elsewhere_marker" not in handler.body, "listed page leaked another allowlisted page's content" + + +def test_wiki_page_read_with_symlinked_root(monkeypatch, tmp_path): + import os as _os + from api import routes + + real_root = tmp_path / "real-wiki" + (real_root / "concepts").mkdir(parents=True) + page = real_root / "concepts" / "real.md" + page.write_text("# real\n", encoding="utf-8") + link_root = tmp_path / "link-wiki" + + try: + _os.symlink(real_root, link_root, target_is_directory=True) + except (OSError, NotImplementedError): + import pytest + pytest.skip("directory symlinks not supported on this platform") + + routes._llm_wiki_clear_page_files_cache() + monkeypatch.setattr(routes, "_llm_wiki_resolve_path", lambda: (link_root, None, None)) + + handler = _FakeHandler() + routes.handle_get(handler, urlparse("http://example.com/api/wiki/page?path=concepts/real.md")) + + assert handler.status == 200, f"listed page should stay readable through symlinked wiki root, got {handler.status}" + assert handler.get_json()["content"] == "# real\n" + + def test_wiki_symlinked_section_cannot_expose_outside_tree(monkeypatch, tmp_path): """A symlinked SECTION dir (concepts -> /tmp/outside) must not expose files outside the real wiki root.""" @@ -273,6 +535,21 @@ def test_wiki_legit_filename_with_dotdot_substring_opens(monkeypatch, tmp_path): assert "diff notes" in h.get_json()["content"] +def test_wiki_page_alias_spellings_are_rejected(monkeypatch, tmp_path): + from api import routes + + wiki_root = tmp_path / "wiki" + (wiki_root / "concepts").mkdir(parents=True) + (wiki_root / "concepts" / "real.md").write_text("# real\n", encoding="utf-8") + + monkeypatch.setattr(routes, "_llm_wiki_resolve_path", lambda: (wiki_root, None, None)) + + for alias in ("concepts/./real.md", "concepts//real.md", "concepts/real.md/", "concepts%5Creal.md"): + handler = _FakeHandler() + routes.handle_get(handler, urlparse(f"http://example.com/api/wiki/page?path={alias}")) + assert handler.status == 400, f"alias spelling {alias!r} must be rejected, got {handler.status}" + + def test_i18n_wiki_keys_in_all_locales(): src = (REPO / "static" / "i18n.js").read_text(encoding="utf-8") required_keys = [ @@ -297,3 +574,92 @@ def test_i18n_wiki_keys_in_all_locales(): f"i18n key '{key}' missing from locale block {i + 1} " f"(position ~{lang_positions[i]})" ) + + +def test_wiki_page_files_documents_hardlink_trust_boundary(): + from api import routes + + doc = routes._llm_wiki_page_files.__doc__ or "" + assert "hardlink" in doc.lower() + assert "trusted" in doc.lower() or "operator-controlled" in doc.lower() + + +def test_wiki_page_files_reuses_cache_with_unchanged_section_mtime(monkeypatch, tmp_path): + from api import routes + + wiki_root = tmp_path / "wiki" + (wiki_root / "concepts").mkdir(parents=True) + page = wiki_root / "concepts" / "one.md" + page.write_text("# one\n", encoding="utf-8") + + routes._llm_wiki_clear_page_files_cache() + monkeypatch.setattr(routes, "_WIKI_ALLOWLIST_TTL", 60.0) + + calls = [] + + def fake_uncached(root): + calls.append(root) + return [page] + + monkeypatch.setattr(routes, "_llm_wiki_page_files_uncached", fake_uncached) + + assert routes._llm_wiki_page_files(wiki_root) == [page] + assert routes._llm_wiki_page_files(wiki_root) == [page] + assert len(calls) == 1 + + +def test_wiki_page_files_cache_invalidates_when_section_mtime_changes(monkeypatch, tmp_path): + import time as _time + from api import routes + + wiki_root = tmp_path / "wiki" + section = wiki_root / "concepts" + section.mkdir(parents=True) + first = section / "one.md" + second = section / "two.md" + first.write_text("# one\n", encoding="utf-8") + + routes._llm_wiki_clear_page_files_cache() + monkeypatch.setattr(routes, "_WIKI_ALLOWLIST_TTL", 60.0) + + calls = [] + + def fake_uncached(root): + calls.append(root) + return [first] if len(calls) == 1 else [first, second] + + monkeypatch.setattr(routes, "_llm_wiki_page_files_uncached", fake_uncached) + + assert routes._llm_wiki_page_files(wiki_root) == [first] + second.write_text("# two\n", encoding="utf-8") + _time.sleep(0.02) # Ensure mtime granularity is sufficient on Windows + os.utime(str(section), None) + assert routes._llm_wiki_page_files(wiki_root) == [first, second] + assert len(calls) == 2 + + +def test_wiki_page_files_cache_expires_after_ttl(monkeypatch, tmp_path): + from api import routes + + wiki_root = tmp_path / "wiki" + (wiki_root / "concepts").mkdir(parents=True) + page = wiki_root / "concepts" / "one.md" + page.write_text("# one\n", encoding="utf-8") + + routes._llm_wiki_clear_page_files_cache() + monkeypatch.setattr(routes, "_WIKI_ALLOWLIST_TTL", 1.0) + + calls = [] + ticks = iter([100.0, 100.5, 101.5]) + monkeypatch.setattr(routes.time, "monotonic", lambda: next(ticks)) + + def fake_uncached(root): + calls.append(root) + return [page] + + monkeypatch.setattr(routes, "_llm_wiki_page_files_uncached", fake_uncached) + + routes._llm_wiki_page_files(wiki_root) # miss → rebuild (t=100.0) + routes._llm_wiki_page_files(wiki_root) # hit (t=100.5 < 101.0) + routes._llm_wiki_page_files(wiki_root) # miss → rebuild (t=101.5 > 101.0) + assert len(calls) == 2 diff --git a/tests/test_stage299_opus_fixes.py b/tests/test_stage299_opus_fixes.py index 8c1aeb64..1f8afb8c 100644 --- a/tests/test_stage299_opus_fixes.py +++ b/tests/test_stage299_opus_fixes.py @@ -42,7 +42,7 @@ def test_count_files_has_iteration_cap(): def test_page_files_has_iteration_cap(): src = _read_source() - start = src.find("def _llm_wiki_page_files(") + start = src.find("def _llm_wiki_page_files_uncached(") end = src.find("\ndef ", start + 1) body = src[start:end] assert "_LLM_WIKI_MAX_FILES" in body