Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions app/api/docs.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()}
12 changes: 9 additions & 3 deletions app/api/system.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
18 changes: 11 additions & 7 deletions app/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
143 changes: 143 additions & 0 deletions tests/test_path_handling.py
Original file line number Diff line number Diff line change
@@ -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("<html>THE-INDEX</html>")
(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())
Loading