handle unsafe user provided paths

Author: rmorshea

requirements/pkg-deps.txt

@@ -1,7 +1,8 @@
 typing-extensions >=3.10
 mypy-extensions >=0.4.3
-anyio >=3.0
+anyio >=3
 jsonpatch >=1.32
 fastjsonschema >=2.14.5
-requests >=2.0
+requests >=2
 colorlog >=6
+werkzeug >=2

src/idom/server/flask.py

@@ -17,22 +17,21 @@
     Request,
     copy_current_request_context,
     request,
-    send_from_directory,
+    send_file,
 )
 from flask_cors import CORS
 from flask_sock import Sock
 from simple_websocket import Server as WebSocket
 from werkzeug.serving import BaseWSGIServer, make_server
 
 import idom
-from idom.config import IDOM_WEB_MODULES_DIR
 from idom.core.hooks import Context, create_context, use_context
 from idom.core.layout import LayoutEvent, LayoutUpdate
 from idom.core.serve import serve_json_patch
 from idom.core.types import ComponentType, RootComponentConstructor
 from idom.utils import Ref
 
-from .utils import CLIENT_BUILD_DIR, client_build_dir_path
+from .utils import safe_client_build_dir_path, safe_web_modules_dir_path
 
 
 logger = logging.getLogger(__name__)
@@ -159,18 +158,15 @@ def _setup_common_routes(blueprint: Blueprint, options: Options) -> None:
         @blueprint.route("/")
         @blueprint.route("/<path:path>")
         def send_client_dir(path: str = "") -> Any:
-            return send_from_directory(
-                str(CLIENT_BUILD_DIR),
-                client_build_dir_path(path),
-            )
+            return send_file(safe_client_build_dir_path(path))
 
         @blueprint.route(r"/_api/modules/<path:path>")
         @blueprint.route(r"<path:_>/_api/modules/<path:path>")
         def send_modules_dir(
             path: str,
             _: str = "",  # this is not used
         ) -> Any:
-            return send_from_directory(str(IDOM_WEB_MODULES_DIR.current), path)
+            return send_file(safe_web_modules_dir_path(path))
 
 
 def _setup_single_view_dispatcher_route(
@@ -258,7 +254,7 @@ def run_send() -> None:
             dispatch_thread_info.dispatch_loop.call_soon_threadsafe(
                 dispatch_thread_info.async_recv_queue.put_nowait, value
             )
-    finally:
+    finally:  # pragma: no cover
         dispatch_thread_info.dispatch_loop.call_soon_threadsafe(
             dispatch_thread_info.dispatch_future.cancel
         )

src/idom/server/sanic.py

@@ -14,7 +14,6 @@
 from sanic_cors import CORS
 from websockets.legacy.protocol import WebSocketCommonProtocol
 
-from idom.config import IDOM_WEB_MODULES_DIR
 from idom.core.hooks import Context, create_context, use_context
 from idom.core.layout import Layout, LayoutEvent
 from idom.core.serve import (
@@ -27,7 +26,7 @@
 from idom.core.types import RootComponentConstructor
 
 from ._asgi import serve_development_asgi
-from .utils import CLIENT_BUILD_DIR, client_build_dir_path
+from .utils import safe_client_build_dir_path, safe_web_modules_dir_path
 
 
 logger = logging.getLogger(__name__)
@@ -122,7 +121,7 @@ async def single_page_app_files(
             path: str = "",
         ) -> response.HTTPResponse:
             path = urllib_parse.unquote(path)
-            return await response.file(CLIENT_BUILD_DIR / client_build_dir_path(path))
+            return await response.file(safe_client_build_dir_path(path))
 
         blueprint.add_route(single_page_app_files, "/")
         blueprint.add_route(single_page_app_files, "/<path:path>")
@@ -132,9 +131,8 @@ async def web_module_files(
             path: str,
             _: str = "",  # this is not used
         ) -> response.HTTPResponse:
-            wm_dir = IDOM_WEB_MODULES_DIR.current
             path = urllib_parse.unquote(path)
-            return await response.file(wm_dir.joinpath(*path.split("/")))
+            return await response.file(safe_web_modules_dir_path(path))
 
         blueprint.add_route(web_module_files, "/_api/modules/<path:path>")
         blueprint.add_route(web_module_files, "/<_:path>/_api/modules/<path:path>")

src/idom/server/starlette.py

@@ -24,7 +24,7 @@
 from idom.core.types import RootComponentConstructor
 
 from ._asgi import serve_development_asgi
-from .utils import CLIENT_BUILD_DIR, client_build_dir_path
+from .utils import CLIENT_BUILD_DIR, safe_client_build_dir_path
 
 
 logger = logging.getLogger(__name__)
@@ -133,11 +133,10 @@ def single_page_app_files() -> Callable[..., Awaitable[None]]:
     )
 
     async def spa_app(scope: Scope, receive: Receive, send: Send) -> None:
-        return await static_files_app(
-            {**scope, "path": client_build_dir_path(scope["path"])},
-            receive,
-            send,
-        )
+        # Path safety is the responsibility of starlette.staticfiles.StaticFiles -
+        # using `safe_client_build_dir_path` is for convenience in this case.
+        path = safe_client_build_dir_path(scope["path"]).name
+        return await static_files_app({**scope, "path": path}, receive, send)
 
     return spa_app
 

src/idom/server/tornado.py

@@ -22,7 +22,7 @@
 from idom.core.serve import VdomJsonPatch, serve_json_patch
 from idom.core.types import ComponentConstructor
 
-from .utils import CLIENT_BUILD_DIR, client_build_dir_path
+from .utils import CLIENT_BUILD_DIR, safe_client_build_dir_path
 
 
 RequestContext: type[Context[HTTPServerRequest | None]] = create_context(
@@ -165,7 +165,9 @@ def _setup_single_view_dispatcher_route(
 
 class SpaStaticFileHandler(StaticFileHandler):
     async def get(self, path: str, include_body: bool = True) -> None:
-        return await super().get(client_build_dir_path(path), include_body)
+        # Path safety is the responsibility of tornado.web.StaticFileHandler -
+        # using `safe_client_build_dir_path` is for convenience in this case.
+        return await super().get(safe_client_build_dir_path(path).name, include_body)
 
 
 class ModelStreamHandler(WebSocketHandler):

src/idom/server/utils.py

@@ -8,7 +8,10 @@
 from pathlib import Path
 from typing import Any, Iterator
 
+from werkzeug.security import safe_join
+
 import idom
+from idom.config import IDOM_WEB_MODULES_DIR
 from idom.types import RootComponentConstructor
 
 from .types import ServerImplementation
@@ -26,12 +29,6 @@
 )
 
 
-def client_build_dir_path(path: str) -> str:
-    start, _, end = path.rpartition("/")
-    file = end or start
-    return file if (CLIENT_BUILD_DIR / file).is_file() else "index.html"
-
-
 def run(
     component: RootComponentConstructor,
     host: str = "127.0.0.1",
@@ -62,6 +59,27 @@ def run(
     )
 
 
+def safe_client_build_dir_path(path: str) -> Path:
+    """Prevent path traversal out of :data:`CLIENT_BUILD_DIR`"""
+    start, _, end = path.rpartition("/")
+    file = end or start
+    final_path = traversal_safe_path(CLIENT_BUILD_DIR, file)
+    return final_path if final_path.is_file() else (CLIENT_BUILD_DIR / "index.html")
+
+
+def safe_web_modules_dir_path(path: str) -> Path:
+    """Prevent path traversal out of :data:`idom.config.IDOM_WEB_MODULES_DIR`"""
+    return traversal_safe_path(IDOM_WEB_MODULES_DIR.current, *path.split("/"))
+
+
+def traversal_safe_path(root: Path, *unsafe_parts: str | Path) -> Path:
+    """Sanitize user given path using ``werkzeug.security.safe_join``"""
+    path = safe_join(str(root.resolve()), *unsafe_parts)
+    if path is None:
+        raise ValueError("Unsafe path")  # pragma: no cover
+    return Path(path)
+
+
 def find_available_port(
     host: str,
     port_min: int = 8000,