diff --git a/api/routes.py b/api/routes.py index 9ca3ed03..87c49415 100644 --- a/api/routes.py +++ b/api/routes.py @@ -6061,9 +6061,12 @@ def _llm_wiki_last_writer( 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 @@ -6102,17 +6105,27 @@ def _llm_wiki_last_writer( log_path = wiki_path / "log.md" if _within_wiki(log_path) and log_path.is_file(): 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})" + log_stat = log_path.stat() + 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_stat.st_dev, log_stat.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 @@ -6146,8 +6159,17 @@ def _build_llm_wiki_status() -> dict: return base allowlisted_entries = _llm_wiki_allowlisted_entries(wiki_path) - page_entries = list(allowlisted_entries.values()) - page_files = [target for target, _ in page_entries] + verified_page_entries: list[tuple[Path, tuple[int, int], os.stat_result]] = [] + for target, identity in allowlisted_entries.values(): + try: + 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 = [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 @@ -6163,7 +6185,7 @@ 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_entries), @@ -7392,11 +7414,13 @@ def handle_get(handler, parsed) -> bool: return bad(handler, "Wiki not configured or directory not found", status=404) allowlisted_entries = _llm_wiki_allowlisted_entries(Path(wiki_root)) pages = [] - for rel_path, (fp, _) in sorted(allowlisted_entries.items(), key=lambda item: item[0].lower()): + for rel_path, (fp, identity) in sorted(allowlisted_entries.items(), key=lambda item: item[0].lower()): try: st = fp.stat() except OSError: continue + 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": diff --git a/tests/test_issue1257_llm_wiki_status.py b/tests/test_issue1257_llm_wiki_status.py index 38672783..d9afb073 100644 --- a/tests/test_issue1257_llm_wiki_status.py +++ b/tests/test_issue1257_llm_wiki_status.py @@ -184,10 +184,39 @@ def test_llm_wiki_status_last_writer_rechecks_identity(monkeypatch, tmp_path): 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_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 7acab9d9..6eb06be0 100644 --- a/tests/test_issue2941_wiki_browse.py +++ b/tests/test_issue2941_wiki_browse.py @@ -86,7 +86,10 @@ def test_wiki_browse_skips_pages_that_disappear_during_listing(monkeypatch, tmp_ monkeypatch.setattr( routes, "_llm_wiki_allowlisted_entries", - lambda root: {"gone.md": (missing, (0, 0)), "ok.md": (ok, (0, 0))}, + lambda root: { + "gone.md": (missing, (0, 0)), + "ok.md": (ok, (ok.stat().st_dev, ok.stat().st_ino)), + }, ) handler = _FakeHandler() @@ -318,6 +321,36 @@ def test_wiki_browse_drops_cached_entry_replaced_by_directory(monkeypatch, tmp_p 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