diff --git a/oxyroute/__init__.py b/oxyroute/__init__.py index 8391f3c..247fc43 100644 --- a/oxyroute/__init__.py +++ b/oxyroute/__init__.py @@ -10,6 +10,7 @@ from oxyroute.router import APIRouter from oxyroute.security_headers import SecurityHeadersConfig from oxyroute.sse import SSEEvent, send_sse, sse_done +from oxyroute.static import StaticFiles from oxyroute.streaming import stream_bytes, stream_jsonl, stream_text __all__ = [ @@ -24,6 +25,7 @@ "Response", "SSEEvent", "SecurityHeadersConfig", + "StaticFiles", "WebSocket", "__version__", "apply_cors", diff --git a/oxyroute/app.py b/oxyroute/app.py index d4dd619..87607d8 100644 --- a/oxyroute/app.py +++ b/oxyroute/app.py @@ -132,6 +132,15 @@ def set_security_headers(self, config: Any | None) -> None: """ self._app.set_security_headers(config) + def mount(self, path: str, app: Any) -> None: + """Mount another application or handler at a specific path prefix.""" + path = path.rstrip("/") + # Mount the exact prefix + self.get(path)(app) + self.get(path + "/")(app) + # Mount all subpaths + self.get(path + "/*path")(app) + def get( self, path: str, diff --git a/oxyroute/router.py b/oxyroute/router.py index 3e10488..7e19805 100644 --- a/oxyroute/router.py +++ b/oxyroute/router.py @@ -67,6 +67,15 @@ def include_router( full = join_path(prefix, rel) self._routes.append((method, full, handler, merged)) + def mount(self, path: str, app: Any) -> None: + """Mount another application or handler at a specific path prefix.""" + path = path.rstrip("/") + # Mount the exact prefix + self.get(path)(app) + self.get(path + "/")(app) + # Mount all subpaths + self.get(path + "/*path")(app) + def get( self, path: str, diff --git a/oxyroute/static.py b/oxyroute/static.py new file mode 100644 index 0000000..ecb7697 --- /dev/null +++ b/oxyroute/static.py @@ -0,0 +1,66 @@ +import mimetypes +import os +from typing import Any + +from oxyroute.exceptions import HTTPException + + +class StaticFiles: + """ + Serve static files from a directory. + + Can be mounted via: + app.mount("/static", StaticFiles("static", html=True)) + """ + + __name__ = "StaticFiles" + + def __init__( + self, + directory: str, + html: bool = False, + max_age: int | None = None, + ) -> None: + self.directory = os.path.abspath(directory) + if not os.path.isdir(self.directory): + raise RuntimeError(f"Directory {directory} does not exist") + self.html = html + self.max_age = max_age + + def __call__(self, protocol: Any, path: str = "") -> Any: + if ".." in path.split("/"): + raise HTTPException(status_code=403, detail="Forbidden") + + file_path = os.path.abspath(os.path.join(self.directory, path.lstrip("/"))) + + if not file_path.startswith(self.directory): + raise HTTPException(status_code=403, detail="Forbidden") + + if not os.path.exists(file_path) or not os.path.isfile(file_path): + if self.html and os.path.isfile(os.path.join(file_path, "index.html")): + file_path = os.path.join(file_path, "index.html") + else: + raise HTTPException(status_code=404, detail="Not Found") + + content_type, _ = mimetypes.guess_type(file_path) + if content_type is None: + content_type = "application/octet-stream" + + headers = [("content-type", content_type)] + if self.max_age is not None: + headers.append(("cache-control", f"public, max-age={self.max_age}")) + + if hasattr(protocol, "response_file"): + # Rust fast path via tokio-fs + protocol.response_file(200, headers, file_path) + from oxyroute.streaming import stream_done + + return stream_done() + else: + # Fallback for Python testing transports without response_file + with open(file_path, "rb") as f: + body = f.read() + protocol.response_bytes(200, headers, body) + from oxyroute.streaming import stream_done + + return stream_done() diff --git a/oxyroute/testing.py b/oxyroute/testing.py index 1a0cba1..1ba44a5 100644 --- a/oxyroute/testing.py +++ b/oxyroute/testing.py @@ -250,6 +250,16 @@ def response_bytes(self, status: int, headers: list, body: bytes) -> None: } ) + def response_file(self, status: int, headers: list, file: str) -> None: + import anyio + + async def _read_file() -> bytes: + async with await anyio.open_file(file, "rb") as f: + return await f.read() + + body = asyncio.run_coroutine_threadsafe(_read_file(), self._loop).result() + self.response_bytes(status, headers, body) + def response_empty(self, status: int, headers: list) -> None: self._status = int(status) self.status = int(status) diff --git a/tests/test_static.py b/tests/test_static.py new file mode 100644 index 0000000..53a6b38 --- /dev/null +++ b/tests/test_static.py @@ -0,0 +1,66 @@ +import os +from tempfile import TemporaryDirectory + +from oxyroute import App, StaticFiles +from oxyroute.testing import TestClient + + +def test_static_files(): + with TemporaryDirectory() as tmpdir: + with open(os.path.join(tmpdir, "test.txt"), "w") as f: + f.write("hello world") + + app = App() + app.mount("/static", StaticFiles(tmpdir)) + + with TestClient(app) as client: + resp = client.get("/static/test.txt") + assert resp.status_code == 200 + assert resp.content == b"hello world" + assert resp.headers["content-type"] == "text/plain" + + +def test_static_files_missing(): + with TemporaryDirectory() as tmpdir: + app = App() + app.mount("/static", StaticFiles(tmpdir)) + + with TestClient(app) as client: + resp = client.get("/static/missing.txt") + assert resp.status_code == 404 + + +def test_static_files_traversal(): + with TemporaryDirectory() as tmpdir: + # Create a file outside the static directory + with open(os.path.join(tmpdir, "secret.txt"), "w") as f: + f.write("secret") + + static_dir = os.path.join(tmpdir, "static") + os.makedirs(static_dir) + + app = App() + app.mount("/static", StaticFiles(static_dir)) + + with TestClient(app) as client: + resp = client.get("/static/../secret.txt") + assert resp.status_code in (403, 404) + + +def test_static_files_index_html(): + with TemporaryDirectory() as tmpdir: + with open(os.path.join(tmpdir, "index.html"), "w") as f: + f.write("

Hello

") + + app = App() + app.mount("/static", StaticFiles(tmpdir, html=True)) + + with TestClient(app) as client: + resp = client.get("/static/") + assert resp.status_code == 200 + assert resp.content == b"

Hello

" + assert resp.headers["content-type"] == "text/html" + + resp = client.get("/static") + assert resp.status_code == 200 + assert resp.content == b"

Hello

"