Merge pull request #4508 from nesquena/release-v0.51.523-4437
Release v0.51.523 — cache trusted wiki page listings + reject hardlinked pages (#4375)
This commit is contained in:
5 files changed
+814
-61
No files matched your search
@@ -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
|
||||
|
||||
+239
-36
@@ -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,20 +6108,33 @@ 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:
|
||||
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.
|
||||
@@ -5972,14 +6149,24 @@ def _llm_wiki_last_writer(wiki_path: Path, page_files: list[Path]) -> str:
|
||||
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:
|
||||
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 == "":
|
||||
@@ -5990,6 +6177,9 @@ def _llm_wiki_last_writer(wiki_path: Path, page_files: list[Path]) -> str:
|
||||
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:
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user