diff --git a/app/api/docs.py b/app/api/docs.py index b4b4067..e7f9c60 100644 --- a/app/api/docs.py +++ b/app/api/docs.py @@ -44,7 +44,9 @@ async def get_doc(slug: str) -> dict: """Return the raw markdown content of one doc, identified by filename stem.""" if not re.fullmatch(r"[A-Za-z0-9_-]+", slug): raise HTTPException(400, "Invalid document identifier") - path = _docs_dir() / f"{slug}.md" - if not path.is_file(): + # The path is taken from the directory listing, and the request's text is + # only compared with it, so nothing in it can ever become part of a path. + path = next((f for f in _docs_dir().glob("*.md") if f.stem == slug), None) + if path is None or not path.is_file(): raise HTTPException(404, "Document not found") return {"slug": slug, "title": _title_from_filename(path.name), "content": path.read_text()} diff --git a/app/api/system.py b/app/api/system.py index 0373691..20a430c 100644 --- a/app/api/system.py +++ b/app/api/system.py @@ -447,14 +447,20 @@ async def restore_from_snapshot(user: AdminUser, snapshot_name: str, files: Opti `files` is an optional comma-separated subset of {pktwifi.db, config.yaml} — omit to restore everything present in the snapshot. """ - if not re.fullmatch(r"backup_\d{4}-\d{2}-\d{2}_\d{2}-\d{2}", snapshot_name): + if not re.fullmatch(r"backup_[0-9]{4}-[0-9]{2}-[0-9]{2}_[0-9]{2}-[0-9]{2}", snapshot_name): raise HTTPException(status_code=400, detail="Invalid snapshot name") settings = get_settings() s = await asyncio.to_thread(_read_backup_settings_sync, settings.db_path) backup_root = Path(s["backup_path"]).resolve() - snap_dir = (backup_root / snapshot_name).resolve() - if snap_dir.parent != backup_root or not snap_dir.is_dir(): + # The directory is taken from the listing of the backup root, and the + # request's text is only compared with it. A symlink is refused: a snapshot + # that points somewhere else is not one of ours. + snap_dir = next( + (d for d in backup_root.iterdir() if d.name == snapshot_name and d.is_dir() and not d.is_symlink()), + None, + ) if backup_root.is_dir() else None + if snap_dir is None: raise HTTPException(status_code=404, detail="Snapshot not found") wanted = _parse_files_param(files) diff --git a/app/main.py b/app/main.py index f1b9c6e..6b74dfb 100644 --- a/app/main.py +++ b/app/main.py @@ -419,13 +419,17 @@ async def serve_spa(request: Request, full_path: str): # which its Python taint tracker doesn't recognise as a sanitizer. _dist_root = os.path.normpath(str(_frontend_dist)) _candidate = os.path.normpath(os.path.join(_dist_root, full_path)) - if not (_candidate == _dist_root or _candidate.startswith(_dist_root + os.sep)): - # Path traversal — this handler is unauthenticated and config.yaml - # sits two levels above dist, so "../../config.yaml" previously - # returned the JWT signing key and the credential encryption key. - raise HTTPException(status_code=404, detail="Not found") - static_file = Path(_candidate) - if static_file.exists() and static_file.is_file(): + static_file = None + # The dist root itself (a request for "/") is not a file; it falls through + # to index.html below. Anything else must lie strictly inside it. + if _candidate != _dist_root: + if not _candidate.startswith(_dist_root + os.sep): + # Path traversal — this handler is unauthenticated and config.yaml + # sits two levels above dist, so "../../config.yaml" previously + # returned the JWT signing key and the credential encryption key. + raise HTTPException(status_code=404, detail="Not found") + static_file = Path(_candidate) + if static_file is not None and static_file.is_file(): return FileResponse(str(static_file)) index = _frontend_dist / "index.html" # index.html names the hashed bundles, so a cached copy pins the browser diff --git a/tests/test_path_handling.py b/tests/test_path_handling.py new file mode 100644 index 0000000..bf681cd --- /dev/null +++ b/tests/test_path_handling.py @@ -0,0 +1,143 @@ +#!/usr/bin/env python3 +""" +Paths built from a request: the in-app docs, snapshot restore, and the +single-page-app file server. + +Standalone script — run from the repo root: + python3 tests/test_path_handling.py + +The app is copied into a temporary tree together with a stand-in for the built +frontend, so that files placed where an escape would land (one and two levels +above the build folder, and a sibling folder that shares its prefix) can be +told apart from files the server is meant to serve. The properties worth +proving: + + * a request cannot name a file outside the folder it is served from — not + with `..`, encoded or double-encoded `..`, backslashes, an absolute path, + a null byte, or a sibling folder whose name starts the same way, + * the files it should serve, it still serves, + * only a snapshot that really is a directory under the backup root can be + restored from, and a symlink standing in for one is refused, + * the docs endpoint serves a document by name and nothing else. +""" +from __future__ import annotations + +import os +import shutil +import sqlite3 +import sys +import tempfile +from pathlib import Path + +from cryptography.fernet import Fernet + +REPO_ROOT = Path(__file__).resolve().parents[1] + +ROOT = Path(tempfile.mkdtemp(prefix="pktwifi-paths-")) +shutil.copytree(REPO_ROOT / "app", ROOT / "app", ignore=shutil.ignore_patterns("__pycache__")) +shutil.copytree(REPO_ROOT / "migrations", ROOT / "migrations") +DIST = ROOT / "frontend" / "dist" +(DIST / "assets").mkdir(parents=True) +(DIST / "index.html").write_text("THE-INDEX") +(DIST / "assets" / "app.js").write_text("console.log('real asset')") +(ROOT / "frontend" / "canary_one_up.txt").write_text("CANARY-ONE-ABOVE-DIST") +(ROOT / "canary_two_up.txt").write_text("CANARY-TWO-ABOVE-DIST") +(ROOT / "frontend" / "dist-evil.txt").write_text("CANARY-SIBLING-OF-DIST") +(ROOT / "docs").mkdir() +(ROOT / "docs" / "guide.md").write_text("# The guide") +(ROOT / "outside_docs.md").write_text("CANARY-OUTSIDE-DOCS") +(ROOT / "config.yaml").write_text( + f"install_dir: {ROOT}\nsecret_key: {'a' * 64}\ncredential_key: {Fernet.generate_key().decode()}\nsuite_token: ''\n") +os.environ["PKTWIFI_CONFIG"] = str(ROOT / "config.yaml") +os.environ["PKTWIFI_INSTALL_DIR"] = str(ROOT) +os.environ["PKTWIFI_ADMIN_PASSWORD"] = "admin-test-password" +sys.path.insert(0, str(ROOT)) + +from fastapi.testclient import TestClient # noqa: E402 + +from app.main import app # noqa: E402 + +FAILURES: list[str] = [] +CANARIES = ("CANARY-ONE-ABOVE-DIST", "CANARY-TWO-ABOVE-DIST", "CANARY-SIBLING-OF-DIST", "CANARY-OUTSIDE-DOCS") + + +def check(label: str, passed: bool, detail: str = "") -> None: + print(f"{'PASS' if passed else 'FAIL'} {label}" + (f" — {detail}" if detail and not passed else "")) + if not passed: + FAILURES.append(label) + + +def main() -> int: + with TestClient(app) as client: + r = client.post("/api/auth/login", json={"username": "admin", "password": "admin-test-password"}) + admin = {"Authorization": f"Bearer {r.json()['access_token']}"} + + print("── the frontend file server ──") + check("a real asset is served", "real asset" in client.get("/assets/app.js").text) + check("the index is served at the root", "THE-INDEX" in client.get("/").text) + check("and for an app route", "THE-INDEX" in client.get("/settings").text) + escapes = [ + "/../../canary_two_up.txt", "/..%2f..%2fcanary_two_up.txt", "/%2e%2e/%2e%2e/canary_two_up.txt", + "/assets/../../canary_two_up.txt", "/assets/..%2f..%2f..%2fcanary_two_up.txt", + "/assets/../../../canary_two_up.txt", "/....//....//canary_two_up.txt", + "/assets/%2e%2e%2f%2e%2e%2f%2e%2e%2fcanary_two_up.txt", "/\\..\\..\\canary_two_up.txt", + "/../canary_one_up.txt", "/..%2fcanary_one_up.txt", "/%2e%2e/canary_one_up.txt", + "/assets/../../canary_one_up.txt", "/../dist-evil.txt", "/..%2fdist-evil.txt", + "/..;/..;/canary_two_up.txt", "/%252e%252e/%252e%252e/canary_two_up.txt", + f"/{ROOT}/canary_two_up.txt", f"//{ROOT}/canary_two_up.txt", f"/%2f{str(ROOT)[1:]}/canary_two_up.txt", + "/%00", "/assets/%00../x", "/.well-known/../../../canary_two_up.txt", "//etc/passwd", "/%2fetc/passwd", + ] + leaked = [p for p in escapes if any(c in client.get(p).text for c in CANARIES)] + check(f"none of {len(escapes)} escape attempts returns a file from outside the build folder", + not leaked, str(leaked)) + check("config.yaml is not reachable", "secret_key" not in client.get("/../../config.yaml").text + and "secret_key" not in client.get("/..%2f..%2fconfig.yaml").text) + + print("\n── in-app docs ──") + check("a document is served by name", "The guide" in client.get("/api/docs-content/guide", headers=admin).json().get("content", "")) + check("the list names it", any(d["slug"] == "guide" for d in client.get("/api/docs-content", headers=admin).json())) + for slug in ("..%2foutside_docs", "guide.md", "guide%00", "../outside_docs", "gu ide", "gui%0ade"): + r = client.get(f"/api/docs-content/{slug}", headers=admin) + check(f"{slug!r} is refused", r.status_code in (400, 404) and not any(c in r.text for c in CANARIES), f"{r.status_code} {r.text[:80]}") + check("an unknown document is a 404", client.get("/api/docs-content/nothing", headers=admin).status_code == 404) + check("sign-in is required", client.get("/api/docs-content/guide").status_code == 401) + + print("\n── restoring from a snapshot ──") + backups = ROOT / "backups" + (backups / "backup_2026-10-05_10-00").mkdir(parents=True) + outside = ROOT / "elsewhere" + outside.mkdir() + (backups / "backup_2026-10-05_11-00").symlink_to(outside, target_is_directory=True) + (backups / "backup_2026-10-05_12-00").write_text("a file, not a directory") + + def restore(name: str): + # files=noop selects nothing, so a successful call changes nothing on disk. + return client.post(f"/api/system/backups/restore/{name}", params={"files": "noop"}, headers=admin) + + r = restore("backup_2026-10-05_10-00") + check("a real snapshot is accepted", r.status_code == 200, f"{r.status_code} {r.text}") + check("a snapshot that does not exist is a 404", restore("backup_2026-10-05_09-00").status_code == 404) + check("a symlink standing in for one is refused", restore("backup_2026-10-05_11-00").status_code == 404) + check("a file standing in for one is refused", restore("backup_2026-10-05_12-00").status_code == 404) + for name in ("..%2f..%2fetc", "backup_2026-10-05_10-00%2f..%2f..", "backup_2026-10-05_10-00%0a", + "backup_٢٠٢٦-١٠-٠٥_١٠-٠٠", "backup_2026-10-05_10-00/", "other"): + # Refused is any answer but success: the name is rejected outright (400/404), + # or, when it carries a decoded slash, matches no restore route at all (405). + r = restore(name) + check(f"{name!r} is refused", r.status_code in (400, 404, 405), str(r.status_code)) + check("a non-admin cannot restore", client.post("/api/system/backups/restore/backup_2026-10-05_10-00").status_code == 401) + + # With no backup root at all. + shutil.rmtree(backups) + check("a missing backup folder is a 404, not a crash", restore("backup_2026-10-05_10-00").status_code == 404) + + print() + if FAILURES: + print(f"{len(FAILURES)} FAILED: {', '.join(FAILURES)}") + return 1 + print("All checks passed.") + return 0 + + +if __name__ == "__main__": + sys.exit(main())