From 37a50c51aa9e9db8d04ca5a53b72230de8232b18 Mon Sep 17 00:00:00 2001 From: ahmad-ajmal Date: Mon, 21 Sep 2026 12:44:05 +0100 Subject: [PATCH] Unauthenticated arbitrary file read --- app/ui_layer/adapters/browser_adapter.py | 50 ++++-- tests/test_browser_static_traversal.py | 209 +++++++++++++++++++++++ 2 files changed, 246 insertions(+), 13 deletions(-) create mode 100644 tests/test_browser_static_traversal.py diff --git a/app/ui_layer/adapters/browser_adapter.py b/app/ui_layer/adapters/browser_adapter.py index 2de74c0a..cad8e48f 100644 --- a/app/ui_layer/adapters/browser_adapter.py +++ b/app/ui_layer/adapters/browser_adapter.py @@ -14,7 +14,7 @@ import uuid from concurrent.futures import ThreadPoolExecutor from datetime import datetime -from pathlib import Path +from pathlib import Path, PurePosixPath, PureWindowsPath from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set from aiohttp.client_exceptions import ClientConnectionResetError @@ -168,6 +168,39 @@ def _with_request_id(message: Dict[str, Any]) -> Dict[str, Any]: from aiohttp import web +def _make_static_or_spa(dist: Path): + """Build the catch-all handler: serve a file from dist/, else index.html. + + Every request path is resolved and must land inside the resolved dist/ + (symlinks included); anything else is a 404, never a file from elsewhere. + """ + from aiohttp import web + + dist_root = dist.resolve() + index_path = dist_root / "index.html" + + async def _static_or_spa(request: web.Request) -> web.StreamResponse: + req_path = request.match_info.get("path", "") + if not req_path: + return web.FileResponse(index_path) + # Refuse drive/root/UNC paths before touching the filesystem: joining + # an absolute path replaces dist/ outright, and resolving //host/share + # makes Windows authenticate to that host. + if PureWindowsPath(req_path).anchor or PurePosixPath(req_path).anchor: + raise web.HTTPNotFound() + try: + file_path = (dist_root / req_path).resolve() + except (OSError, ValueError): + raise web.HTTPNotFound() + if not file_path.is_relative_to(dist_root): + raise web.HTTPNotFound() + if file_path.is_file(): + return web.FileResponse(file_path) + return web.FileResponse(index_path) + + return _static_or_spa + + class BrowserThemeAdapter(ThemeAdapter): """Browser-specific theme adapter outputting CSS-compatible styles.""" @@ -1096,19 +1129,10 @@ async def _on_start(self) -> None: # Serve static files from dist/ (public/ files copied by Vite build) # This must come before the SPA catch-all so images, fonts, etc. are served directly - _dist = frontend_dist # capture for closure - - async def _static_or_spa(request: web.Request) -> web.StreamResponse: - """Serve static file from dist/ if it exists, otherwise index.html for SPA routing.""" - req_path = request.match_info.get("path", "") - if req_path: - file_path = _dist / req_path - if file_path.is_file(): - return web.FileResponse(file_path) - return web.FileResponse(_dist / "index.html") - self._app.router.add_get("/", self._spa_handler) - self._app.router.add_get("/{path:.*}", _static_or_spa) + self._app.router.add_get( + "/{path:.*}", _make_static_or_spa(frontend_dist) + ) else: # Fallback to inline HTML for development without build self._app.router.add_get("/", self._index_handler) diff --git a/tests/test_browser_static_traversal.py b/tests/test_browser_static_traversal.py new file mode 100644 index 00000000..8cf64ea9 --- /dev/null +++ b/tests/test_browser_static_traversal.py @@ -0,0 +1,209 @@ +"""Path traversal through the browser adapter's dist/ catch-all route. + +Runs the production handler (_make_static_or_spa) on a real aiohttp server and +sends raw request lines over a socket, so no client normalizes the path first. +""" + +import asyncio +import socket +import sys +import threading +from pathlib import Path + +import pytest +from aiohttp import web + +from app.ui_layer.adapters.browser_adapter import _make_static_or_spa + +SECRET = "SECRET_API_KEY=sk-test-leak" +INDEX = "INDEX" + + +def _can_symlink(tmp: Path) -> bool: + probe = tmp / "_symlink_probe" + try: + probe.symlink_to(tmp) + except (OSError, NotImplementedError): + return False + probe.unlink() + return True + + +@pytest.fixture(scope="module") +def server(tmp_path_factory): + root = tmp_path_factory.mktemp("traversal") + dist = root / "dist" + (dist / "assets").mkdir(parents=True) + (dist / "中文").mkdir() + (dist / "index.html").write_text(INDEX, encoding="utf-8") + (dist / "assets" / "app.js").write_text("console.log('app')", encoding="utf-8") + (dist / "日本語.txt").write_text("japanese-ok", encoding="utf-8") + (dist / "中文" / "图片.txt").write_text("chinese-ok", encoding="utf-8") + (root / ".env").write_text(SECRET, encoding="utf-8") + # A sibling whose name starts with "dist" catches string-prefix checks. + (root / "dist-evil").mkdir() + (root / "dist-evil" / "leak.txt").write_text(SECRET, encoding="utf-8") + + symlinks = _can_symlink(root) + if symlinks: + (dist / "escape.txt").symlink_to(root / ".env") + (dist / "escape_dir").symlink_to(root, target_is_directory=True) + (dist / "alias.js").symlink_to(dist / "assets" / "app.js") + junction = False + if not symlinks and sys.platform == "win32": + # Junctions need no privilege and resolve() follows them like symlinks. + import _winapi + + _winapi.CreateJunction(str(root), str(dist / "escape_junction")) + junction = True + + loop = asyncio.new_event_loop() + ready = threading.Event() + state = {} + + async def start(): + app = web.Application() + app.router.add_get("/{path:.*}", _make_static_or_spa(dist)) + runner = web.AppRunner(app) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + state["runner"] = runner + state["port"] = site._server.sockets[0].getsockname()[1] + ready.set() + + thread = threading.Thread(target=loop.run_forever, daemon=True) + thread.start() + asyncio.run_coroutine_threadsafe(start(), loop) + assert ready.wait(10) + + yield {"port": state["port"], "root": root, "symlinks": symlinks, "junction": junction} + + asyncio.run_coroutine_threadsafe(state["runner"].cleanup(), loop).result(10) + loop.call_soon_threadsafe(loop.stop) + thread.join(10) + + +def _get(server, target: str): + """Send `GET ` verbatim; return (status, body).""" + with socket.create_connection(("127.0.0.1", server["port"]), timeout=10) as s: + s.sendall( + b"GET " + target.encode("utf-8") + b" HTTP/1.1\r\n" + b"Host: localhost\r\nConnection: close\r\n\r\n" + ) + data = b"" + while chunk := s.recv(65536): + data += chunk + head, _, body = data.partition(b"\r\n\r\n") + status = int(head.split(b" ", 2)[1]) + return status, body.decode("utf-8", errors="replace") + + +def _assert_refused(server, target: str): + status, body = _get(server, target) + assert "sk-test-leak" not in body, f"{target} leaked the secret" + assert status == 404, f"{target} -> {status}" + + +# ── Traversal: must be refused ────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "target", + [ + "/../.env", + "/assets/../../.env", + "/%2e%2e/.env", + "/%2e%2e%2f.env", + "/%2E%2E%2F.env", + "/assets%2f%2e%2e%2f%2e%2e%2f.env", + "/../dist-evil/leak.txt", + "/a%00b", + ], +) +def test_relative_traversal_refused(server, target): + _assert_refused(server, target) + + +@pytest.mark.parametrize( + "target", ["/..%5c.env", "/..\\.env", "/assets\\..\\..\\.env"] +) +def test_backslash_traversal_refused(server, target): + if sys.platform == "win32": + _assert_refused(server, target) + else: + # "\" is an ordinary filename character on POSIX: a missing file in dist/. + assert _get(server, target) == (200, INDEX) + + +def test_absolute_windows_paths_refused(server): + secret = server["root"] / ".env" + _assert_refused(server, "/" + secret.as_posix()) # /C:/Users/.../.env + _assert_refused(server, "/" + str(secret).replace("\\", "%5c")) + _assert_refused(server, "/C:%5cWindows%5cwin.ini") + _assert_refused(server, "/C:\\Windows\\win.ini") + _assert_refused(server, "/C:win.ini") # drive-relative + + +def test_absolute_posix_and_unc_paths_refused(server): + _assert_refused(server, "//etc/passwd") + _assert_refused(server, "/%2fetc%2fpasswd") + _assert_refused(server, "/%5cetc%5cpasswd") + _assert_refused(server, "//attacker.invalid/share/x") + _assert_refused(server, "/%5c%5cattacker.invalid%5cshare%5cx") + + +def test_double_encoded_is_not_decoded_twice(server): + status, body = _get(server, "/%252e%252e%252f.env") + assert status == 200 and body == INDEX + + +def test_symlinks_out_of_dist_refused(server): + if not server["symlinks"]: + pytest.skip("symlink creation not permitted (Windows needs Developer Mode)") + _assert_refused(server, "/escape.txt") + _assert_refused(server, "/escape_dir/.env") + + +def test_junction_out_of_dist_refused(server): + if not server["junction"]: + pytest.skip("junction fallback only used on Windows without symlink privilege") + _assert_refused(server, "/escape_junction/.env") + _assert_refused(server, "/escape_junction/dist-evil/leak.txt") + + +def test_symlink_inside_dist_served(server): + if not server["symlinks"]: + pytest.skip("symlink creation not permitted (Windows needs Developer Mode)") + assert _get(server, "/alias.js") == (200, "console.log('app')") + + +# ── Legitimate requests: must still work ──────────────────────────────────── + + +def test_normal_asset_served(server): + assert _get(server, "/assets/app.js") == (200, "console.log('app')") + + +def test_root_file_served(server): + assert _get(server, "/index.html") == (200, INDEX) + + +@pytest.mark.parametrize( + "target, body", + [ + ("/%E6%97%A5%E6%9C%AC%E8%AA%9E.txt", "japanese-ok"), + ("/%E4%B8%AD%E6%96%87/%E5%9B%BE%E7%89%87.txt", "chinese-ok"), + ], +) +def test_non_ascii_filenames_served(server, target, body): + # Browsers percent-encode UTF-8; aiohttp rejects raw non-ASCII request lines. + assert _get(server, target) == (200, body) + + +@pytest.mark.parametrize( + "target", + ["/", "/chat", "/chat/123", "/settings/profile?tab=a", "/assets", "/%E4%B8%AD%E6%96%87"], +) +def test_spa_routes_fall_back_to_index(server, target): + assert _get(server, target) == (200, INDEX)